120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, ConfigDict
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import require_auth
|
|
from app.db import get_session
|
|
from app.models.tables import Category, Transaction
|
|
from app.parsers.validate import dedup_hash
|
|
from app.services.categorize import apply_rules
|
|
|
|
router = APIRouter(prefix="/api/transactions", tags=["transactions"],
|
|
dependencies=[Depends(require_auth)])
|
|
|
|
|
|
class TransactionIn(BaseModel):
|
|
account_id: int
|
|
booking_date: date
|
|
value_date: date | None = None
|
|
amount: Decimal
|
|
purpose: str = ""
|
|
counterparty: str = ""
|
|
force: bool = False
|
|
|
|
|
|
class TransactionOut(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
id: int
|
|
account_id: int
|
|
statement_id: int | None = None
|
|
booking_date: date
|
|
value_date: date | None = None
|
|
amount: Decimal
|
|
purpose: str
|
|
counterparty: str
|
|
category_id: int | None = None
|
|
status: str
|
|
dedup_hash: str
|
|
is_duplicate: bool
|
|
|
|
|
|
class TransactionPatch(BaseModel):
|
|
category_id: int | None = None
|
|
|
|
|
|
@router.get("", response_model=list[TransactionOut])
|
|
def list_transactions(
|
|
account_id: int | None = None,
|
|
date_from: date | None = None,
|
|
date_to: date | None = None,
|
|
category_id: int | None = None,
|
|
q: str | None = None,
|
|
status: str = "confirmed",
|
|
limit: int = 200,
|
|
offset: int = 0,
|
|
session: Session = Depends(get_session),
|
|
):
|
|
stmt = select(Transaction)
|
|
if status:
|
|
stmt = stmt.where(Transaction.status == status)
|
|
if account_id is not None:
|
|
stmt = stmt.where(Transaction.account_id == account_id)
|
|
if date_from is not None:
|
|
stmt = stmt.where(Transaction.booking_date >= date_from)
|
|
if date_to is not None:
|
|
stmt = stmt.where(Transaction.booking_date <= date_to)
|
|
if category_id is not None:
|
|
stmt = stmt.where(Transaction.category_id == category_id)
|
|
if q:
|
|
like = f"%{q}%"
|
|
stmt = stmt.where(or_(Transaction.purpose.ilike(like),
|
|
Transaction.counterparty.ilike(like)))
|
|
stmt = stmt.order_by(Transaction.booking_date, Transaction.id).offset(offset).limit(limit)
|
|
return [TransactionOut.model_validate(t) for t in session.execute(stmt).scalars()]
|
|
|
|
|
|
@router.post("", response_model=TransactionOut, status_code=201)
|
|
def create_transaction(data: TransactionIn, session: Session = Depends(get_session)):
|
|
h = dedup_hash(data.account_id, data.booking_date, data.amount, data.purpose)
|
|
existing = session.execute(
|
|
select(Transaction).where(Transaction.dedup_hash == h)).scalar()
|
|
if existing is not None and not data.force:
|
|
raise HTTPException(409, "Buchung existiert bereits (Duplikat)")
|
|
tx = Transaction(
|
|
account_id=data.account_id,
|
|
booking_date=data.booking_date,
|
|
value_date=data.value_date,
|
|
amount=data.amount,
|
|
purpose=data.purpose,
|
|
counterparty=data.counterparty,
|
|
status="confirmed",
|
|
dedup_hash=h,
|
|
is_duplicate=existing is not None,
|
|
)
|
|
session.add(tx)
|
|
session.commit()
|
|
session.refresh(tx)
|
|
apply_rules(session, [tx])
|
|
session.commit()
|
|
session.refresh(tx)
|
|
return TransactionOut.model_validate(tx)
|
|
|
|
|
|
@router.patch("/{tx_id}", response_model=TransactionOut)
|
|
def patch_transaction(tx_id: int, data: TransactionPatch,
|
|
session: Session = Depends(get_session)):
|
|
tx = session.get(Transaction, tx_id)
|
|
if tx is None:
|
|
raise HTTPException(404, "Buchung nicht gefunden")
|
|
if "category_id" in data.model_fields_set:
|
|
if data.category_id is not None and session.get(Category, data.category_id) is None:
|
|
raise HTTPException(404, "Kategorie nicht gefunden")
|
|
tx.category_id = data.category_id
|
|
session.commit()
|
|
session.refresh(tx)
|
|
return TransactionOut.model_validate(tx)
|