164 lines
6.3 KiB
Python
164 lines
6.3 KiB
Python
from datetime import datetime
|
|
from decimal import Decimal
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy.orm import Session
|
|
from app.auth.dependencies import get_current_user
|
|
from app.database import get_db
|
|
from app.models.account import Account
|
|
from app.models.operation import Operation, OperationType, OperationRelation
|
|
from app.models.user import User
|
|
from app.schemas.operation import (
|
|
OperationCreate, OperationUpdate, OperationOut,
|
|
OperationTypeOut, TransferCreate,
|
|
OperationRelationOut,
|
|
)
|
|
|
|
router = APIRouter(prefix="/operations", tags=["operations"])
|
|
|
|
|
|
def _assert_account_ownership(db: Session, account_id: int, user_id: int) -> Account:
|
|
account = db.get(Account, account_id)
|
|
if not account or account.user_id != user_id:
|
|
raise HTTPException(status_code=404, detail="Account not found")
|
|
return account
|
|
|
|
|
|
def _apply_balance(account: Account, op_type_name: str, amount: Decimal, sign: int = 1):
|
|
if op_type_name == "income":
|
|
account.balance += amount * sign
|
|
elif op_type_name == "expense":
|
|
account.balance -= amount * sign
|
|
|
|
|
|
@router.get("/types", response_model=list[OperationTypeOut])
|
|
def list_types(db: Session = Depends(get_db), _=Depends(get_current_user)):
|
|
return db.query(OperationType).all()
|
|
|
|
|
|
@router.get("", response_model=list[OperationOut])
|
|
def list_operations(
|
|
account_id: int | None = None,
|
|
category_id: int | None = None,
|
|
operation_type_id: int | None = None,
|
|
date_from: datetime | None = None,
|
|
date_to: datetime | None = None,
|
|
is_planned: bool | None = None,
|
|
limit: int = Query(100, le=1000),
|
|
offset: int = 0,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
q = db.query(Operation).filter(Operation.user_id == current_user.id)
|
|
if account_id:
|
|
q = q.filter(Operation.account_id == account_id)
|
|
if category_id:
|
|
q = q.filter(Operation.category_id == category_id)
|
|
if operation_type_id:
|
|
q = q.filter(Operation.operation_type_id == operation_type_id)
|
|
if date_from:
|
|
q = q.filter(Operation.date >= date_from)
|
|
if date_to:
|
|
q = q.filter(Operation.date <= date_to)
|
|
if is_planned is not None:
|
|
q = q.filter(Operation.is_planned == is_planned)
|
|
return q.order_by(Operation.date.desc()).offset(offset).limit(limit).all()
|
|
|
|
|
|
@router.get("/{operation_id}", response_model=OperationOut)
|
|
def get_operation(operation_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
|
obj = db.get(Operation, operation_id)
|
|
if not obj or obj.user_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Operation not found")
|
|
return obj
|
|
|
|
|
|
@router.post("", response_model=OperationOut, status_code=status.HTTP_201_CREATED)
|
|
def create_operation(
|
|
body: OperationCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
|
|
):
|
|
account = _assert_account_ownership(db, body.account_id, current_user.id)
|
|
op_type = db.get(OperationType, body.operation_type_id)
|
|
if not op_type:
|
|
raise HTTPException(status_code=400, detail="Invalid operation type")
|
|
obj = Operation(**body.model_dump(), user_id=current_user.id)
|
|
db.add(obj)
|
|
_apply_balance(account, op_type.name, body.amount)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return obj
|
|
|
|
|
|
@router.post("/transfer", response_model=list[OperationOut], status_code=status.HTTP_201_CREATED)
|
|
def create_transfer(
|
|
body: TransferCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
|
|
):
|
|
from_account = _assert_account_ownership(db, body.from_account_id, current_user.id)
|
|
to_account = _assert_account_ownership(db, body.to_account_id, current_user.id)
|
|
transfer_type = db.query(OperationType).filter(OperationType.name == "transfer").first()
|
|
if not transfer_type:
|
|
raise HTTPException(status_code=500, detail="Transfer operation type not configured")
|
|
|
|
op_out = Operation(
|
|
user_id=current_user.id, account_id=body.from_account_id,
|
|
operation_type_id=transfer_type.id, amount=body.amount,
|
|
currency_id=body.currency_id, exchange_rate=body.exchange_rate,
|
|
amount_converted=body.amount_converted, date=body.date,
|
|
description=body.description, location_id=body.location_id,
|
|
)
|
|
amount_in = body.amount_converted or body.amount
|
|
op_in = Operation(
|
|
user_id=current_user.id, account_id=body.to_account_id,
|
|
operation_type_id=transfer_type.id, amount=amount_in,
|
|
currency_id=to_account.currency_id, date=body.date,
|
|
description=body.description, location_id=body.location_id,
|
|
)
|
|
from_account.balance -= body.amount
|
|
to_account.balance += amount_in
|
|
db.add_all([op_out, op_in])
|
|
db.flush()
|
|
|
|
relation = OperationRelation(
|
|
operation_from_id=op_out.id, operation_to_id=op_in.id, relation_type="transfer"
|
|
)
|
|
db.add(relation)
|
|
db.commit()
|
|
db.refresh(op_out)
|
|
db.refresh(op_in)
|
|
return [op_out, op_in]
|
|
|
|
|
|
@router.patch("/{operation_id}", response_model=OperationOut)
|
|
def update_operation(
|
|
operation_id: int, body: OperationUpdate,
|
|
db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
|
|
):
|
|
obj = db.get(Operation, operation_id)
|
|
if not obj or obj.user_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Operation not found")
|
|
|
|
if body.amount is not None and body.amount != obj.amount:
|
|
account = db.get(Account, obj.account_id)
|
|
op_type = db.get(OperationType, obj.operation_type_id)
|
|
_apply_balance(account, op_type.name, obj.amount, sign=-1)
|
|
_apply_balance(account, op_type.name, body.amount, sign=1)
|
|
|
|
for field, value in body.model_dump(exclude_none=True).items():
|
|
setattr(obj, field, value)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return obj
|
|
|
|
|
|
@router.delete("/{operation_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_operation(
|
|
operation_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
|
|
):
|
|
obj = db.get(Operation, operation_id)
|
|
if not obj or obj.user_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Operation not found")
|
|
account = db.get(Account, obj.account_id)
|
|
op_type = db.get(OperationType, obj.operation_type_id)
|
|
_apply_balance(account, op_type.name, obj.amount, sign=-1)
|
|
db.delete(obj)
|
|
db.commit()
|