81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
from app.auth.dependencies import get_current_user
|
|
from app.database import get_db
|
|
from app.models.budget import Budget
|
|
from app.models.operation import Operation
|
|
from app.models.user import User
|
|
from app.schemas.budget import BudgetCreate, BudgetUpdate, BudgetOut
|
|
|
|
router = APIRouter(prefix="/budgets", tags=["budgets"])
|
|
|
|
|
|
def _compute_spent(db: Session, budget: Budget, current_user: User) -> Decimal:
|
|
period_start = budget.start_date
|
|
period_end = budget.end_date or date.today()
|
|
total = db.query(func.sum(Operation.amount)).filter(
|
|
Operation.user_id == current_user.id,
|
|
Operation.category_id == budget.category_id,
|
|
func.date(Operation.date) >= period_start,
|
|
func.date(Operation.date) <= period_end,
|
|
Operation.is_planned == False,
|
|
).scalar() or Decimal("0")
|
|
return total
|
|
|
|
|
|
@router.get("", response_model=list[BudgetOut])
|
|
def list_budgets(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
|
budgets = db.query(Budget).filter(Budget.user_id == current_user.id).all()
|
|
result = []
|
|
for b in budgets:
|
|
out = BudgetOut.model_validate(b)
|
|
out.spent = _compute_spent(db, b, current_user)
|
|
result.append(out)
|
|
return result
|
|
|
|
|
|
@router.get("/{budget_id}", response_model=BudgetOut)
|
|
def get_budget(budget_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
|
obj = db.get(Budget, budget_id)
|
|
if not obj or obj.user_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Budget not found")
|
|
out = BudgetOut.model_validate(obj)
|
|
out.spent = _compute_spent(db, obj, current_user)
|
|
return out
|
|
|
|
|
|
@router.post("", response_model=BudgetOut, status_code=status.HTTP_201_CREATED)
|
|
def create_budget(body: BudgetCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
|
obj = Budget(**body.model_dump(), user_id=current_user.id)
|
|
db.add(obj)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return BudgetOut.model_validate(obj)
|
|
|
|
|
|
@router.patch("/{budget_id}", response_model=BudgetOut)
|
|
def update_budget(
|
|
budget_id: int, body: BudgetUpdate,
|
|
db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
|
|
):
|
|
obj = db.get(Budget, budget_id)
|
|
if not obj or obj.user_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Budget not found")
|
|
for field, value in body.model_dump(exclude_none=True).items():
|
|
setattr(obj, field, value)
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return BudgetOut.model_validate(obj)
|
|
|
|
|
|
@router.delete("/{budget_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_budget(budget_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
|
obj = db.get(Budget, budget_id)
|
|
if not obj or obj.user_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Budget not found")
|
|
db.delete(obj)
|
|
db.commit()
|