103 lines
3.3 KiB
Python
103 lines
3.3 KiB
Python
import re
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.responses import JSONResponse
|
|
from sqlmodel import Session, select
|
|
|
|
from ..database import get_session
|
|
from ..deps import get_current_user
|
|
from ..models import Budget, User
|
|
from ..schemas.budget import BudgetCreate, BudgetResponse, BudgetUpdate
|
|
|
|
router = APIRouter(prefix="/api/budgets", tags=["budgets"])
|
|
|
|
HEX_COLOR = re.compile(r"^#[0-9a-fA-F]{6}$")
|
|
|
|
|
|
def _serialize(budget: Budget) -> dict:
|
|
return BudgetResponse.model_validate(budget).model_dump(by_alias=True)
|
|
|
|
|
|
def _validate_color(color: str | None) -> None:
|
|
if color and not HEX_COLOR.match(color):
|
|
raise HTTPException(status_code=400, detail="color must be a 6-digit hex string e.g. #6366f1")
|
|
|
|
|
|
@router.get("")
|
|
def list_budgets(
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
budgets = session.exec(
|
|
select(Budget).where(Budget.user_id == current_user.id).order_by(Budget.created_at)
|
|
).all()
|
|
return JSONResponse(content=jsonable_encoder([_serialize(b) for b in budgets]))
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
def create_budget(
|
|
body: BudgetCreate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
_validate_color(body.color)
|
|
if body.limit_cents is not None and body.limit_cents <= 0:
|
|
raise HTTPException(status_code=400, detail="limitCents must be positive")
|
|
|
|
budget = Budget(
|
|
user_id=current_user.id,
|
|
name=body.name,
|
|
limit_cents=body.limit_cents,
|
|
color=body.color,
|
|
)
|
|
session.add(budget)
|
|
session.commit()
|
|
session.refresh(budget)
|
|
return JSONResponse(content=jsonable_encoder(_serialize(budget)), status_code=201)
|
|
|
|
|
|
@router.patch("/{budget_id}")
|
|
def update_budget(
|
|
budget_id: str,
|
|
body: BudgetUpdate,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
budget = session.exec(
|
|
select(Budget).where(Budget.id == budget_id, Budget.user_id == current_user.id)
|
|
).first()
|
|
if not budget:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
|
|
updates = body.model_dump(exclude_unset=True, by_alias=False)
|
|
if "color" in updates:
|
|
_validate_color(updates["color"])
|
|
if "limit_cents" in updates and updates["limit_cents"] is not None and updates["limit_cents"] <= 0:
|
|
raise HTTPException(status_code=400, detail="limitCents must be positive")
|
|
|
|
for field, value in updates.items():
|
|
setattr(budget, field, value)
|
|
budget.updated_at = datetime.utcnow()
|
|
session.add(budget)
|
|
session.commit()
|
|
session.refresh(budget)
|
|
return JSONResponse(content=jsonable_encoder(_serialize(budget)))
|
|
|
|
|
|
@router.delete("/{budget_id}", status_code=204)
|
|
def delete_budget(
|
|
budget_id: str,
|
|
session: Session = Depends(get_session),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
budget = session.exec(
|
|
select(Budget).where(Budget.id == budget_id, Budget.user_id == current_user.id)
|
|
).first()
|
|
if not budget:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
# ON DELETE SET NULL on Transaction.budgetId handled by the DB FK
|
|
session.delete(budget)
|
|
session.commit()
|