158 lines
6.4 KiB
Python
158 lines
6.4 KiB
Python
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()
|