from typing import List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from ..database import get_db from .. import models, schemas router = APIRouter(prefix="/api", tags=["piggy-banks"]) # --- Categories --- @router.get("/categories", response_model=List[schemas.CategoryResponse]) def list_categories(db: Session = Depends(get_db)): return db.query(models.Category).order_by(models.Category.name).all() @router.post("/categories", response_model=schemas.CategoryResponse, status_code=status.HTTP_201_CREATED) def create_category(payload: schemas.CategoryCreate, db: Session = Depends(get_db)): existing = db.query(models.Category).filter(models.Category.name == payload.name).first() if existing: raise HTTPException(status_code=400, detail="Category already exists") category = models.Category(name=payload.name) db.add(category) db.commit() db.refresh(category) return category # --- Piggy Banks --- @router.get("/piggy-banks", response_model=List[schemas.PiggyBankResponse]) def list_piggy_banks(db: Session = Depends(get_db)): banks = db.query(models.PiggyBank).order_by(models.PiggyBank.created_at.desc()).all() result = [] for bank in banks: progress = round((bank.current_amount / bank.target_amount) * 100, 2) if bank.target_amount > 0 else 0 result.append(schemas.PiggyBankResponse( id=bank.id, name=bank.name, target_amount=bank.target_amount, current_amount=bank.current_amount, currency=bank.currency, category_id=bank.category_id, progress_percent=progress, created_at=bank.created_at, updated_at=bank.updated_at, )) return result @router.get("/piggy-banks/{bank_id}", response_model=schemas.PiggyBankResponse) def get_piggy_bank(bank_id: int, db: Session = Depends(get_db)): bank = db.query(models.PiggyBank).filter(models.PiggyBank.id == bank_id).first() if not bank: raise HTTPException(status_code=404, detail="Piggy bank not found") progress = round((bank.current_amount / bank.target_amount) * 100, 2) if bank.target_amount > 0 else 0 return schemas.PiggyBankResponse( id=bank.id, name=bank.name, target_amount=bank.target_amount, current_amount=bank.current_amount, currency=bank.currency, category_id=bank.category_id, progress_percent=progress, created_at=bank.created_at, updated_at=bank.updated_at, ) @router.post("/piggy-banks", response_model=schemas.PiggyBankResponse, status_code=status.HTTP_201_CREATED) def create_piggy_bank(payload: schemas.PiggyBankCreate, db: Session = Depends(get_db)): if payload.category_id: cat = db.query(models.Category).filter(models.Category.id == payload.category_id).first() if not cat: raise HTTPException(status_code=404, detail="Category not found") bank = models.PiggyBank(**payload.model_dump()) db.add(bank) db.commit() db.refresh(bank) progress = round((bank.current_amount / bank.target_amount) * 100, 2) if bank.target_amount > 0 else 0 return schemas.PiggyBankResponse( id=bank.id, name=bank.name, target_amount=bank.target_amount, current_amount=bank.current_amount, currency=bank.currency, category_id=bank.category_id, progress_percent=progress, created_at=bank.created_at, updated_at=bank.updated_at, ) @router.patch("/piggy-banks/{bank_id}", response_model=schemas.PiggyBankResponse) def update_piggy_bank(bank_id: int, payload: schemas.PiggyBankUpdate, db: Session = Depends(get_db)): bank = db.query(models.PiggyBank).filter(models.PiggyBank.id == bank_id).first() if not bank: raise HTTPException(status_code=404, detail="Piggy bank not found") update_data = payload.model_dump(exclude_unset=True) if "category_id" in update_data and update_data["category_id"] is not None: cat = db.query(models.Category).filter(models.Category.id == update_data["category_id"]).first() if not cat: raise HTTPException(status_code=404, detail="Category not found") for key, value in update_data.items(): setattr(bank, key, value) db.commit() db.refresh(bank) progress = round((bank.current_amount / bank.target_amount) * 100, 2) if bank.target_amount > 0 else 0 return schemas.PiggyBankResponse( id=bank.id, name=bank.name, target_amount=bank.target_amount, current_amount=bank.current_amount, currency=bank.currency, category_id=bank.category_id, progress_percent=progress, created_at=bank.created_at, updated_at=bank.updated_at, ) @router.delete("/piggy-banks/{bank_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_piggy_bank(bank_id: int, db: Session = Depends(get_db)): bank = db.query(models.PiggyBank).filter(models.PiggyBank.id == bank_id).first() if not bank: raise HTTPException(status_code=404, detail="Piggy bank not found") db.delete(bank) db.commit() # --- Transactions --- @router.post("/piggy-banks/{bank_id}/transactions", response_model=schemas.TransactionResponse, status_code=status.HTTP_201_CREATED) def create_transaction(bank_id: int, payload: schemas.TransactionCreate, db: Session = Depends(get_db)): bank = db.query(models.PiggyBank).filter(models.PiggyBank.id == bank_id).first() if not bank: raise HTTPException(status_code=404, detail="Piggy bank not found") if bank.current_amount + payload.amount < 0: raise HTTPException(status_code=400, detail="Insufficient funds") bank.current_amount += payload.amount transaction = models.Transaction( piggy_bank_id=bank_id, amount=payload.amount, description=payload.description, ) db.add(transaction) db.commit() db.refresh(transaction) return transaction @router.get("/piggy-banks/{bank_id}/transactions", response_model=List[schemas.TransactionResponse]) def list_transactions(bank_id: int, db: Session = Depends(get_db)): bank = db.query(models.PiggyBank).filter(models.PiggyBank.id == bank_id).first() if not bank: raise HTTPException(status_code=404, detail="Piggy bank not found") return db.query(models.Transaction).filter( models.Transaction.piggy_bank_id == bank_id ).order_by(models.Transaction.created_at.desc()).all()