File size: 4,038 Bytes
fade1d6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from fastapi import HTTPException, status
from bson import ObjectId
from ..database.connection import get_collection
from ..models.monthly_record import (
MonthlyRecord, MonthlyRecordCreate, MonthlyRecordUpdate,
ExpenseCategory, month_key_from, normalize_month_name
)
class RecordService:
def __init__(self):
self.col = get_collection()
async def ensure_indexes(self) -> None:
# Unique index on month_key for quick lookup
await self.col.create_index("month_key", unique=True)
def _calc_totals(self, salary: float, expenses: List[ExpenseCategory]) -> Dict[str, float]:
total_expenses = round(sum(e.amount for e in expenses), 2)
remaining = round(salary - total_expenses, 2)
return {"total_expenses": total_expenses, "remaining": remaining}
def _serialize(self, doc: Dict[str, Any]) -> Dict[str, Any]:
if not doc:
return doc
doc["_id"] = str(doc["_id"])
return doc
async def get_by_month_key(self, month_key: str) -> Dict[str, Any]:
doc = await self.col.find_one({"month_key": month_key})
if not doc:
raise HTTPException(status_code=404, detail="No record found for this month")
return self._serialize(doc)
async def list(self, limit: int = 12, skip: int = 0) -> List[Dict[str, Any]]:
cursor = self.col.find({}).sort("month_key", 1).skip(skip).limit(limit)
return [self._serialize(d) async for d in cursor]
async def create(self, payload: MonthlyRecordCreate) -> Dict[str, Any]:
month = normalize_month_name(payload.month)
month_key = month_key_from(month, payload.year)
now = datetime.now(timezone.utc)
totals = self._calc_totals(payload.salary, payload.expenses)
doc = {
"month": month,
"year": payload.year,
"month_key": month_key,
"salary": float(payload.salary),
"expenses": [e.model_dump() for e in payload.expenses],
"total_expenses": totals["total_expenses"],
"remaining": totals["remaining"],
"created_at": now,
"updated_at": now,
}
try:
result = await self.col.insert_one(doc)
except Exception as e:
# likely duplicate month_key
raise HTTPException(status_code=400, detail=str(e))
created = await self.col.find_one({"_id": result.inserted_id})
return self._serialize(created)
async def update(self, month_key: str, payload: MonthlyRecordUpdate) -> Dict[str, Any]:
existing = await self.col.find_one({"month_key": month_key})
if not existing:
raise HTTPException(status_code=404, detail="No record found for this month")
salary = payload.salary if payload.salary is not None else existing["salary"]
expenses = payload.expenses if payload.expenses is not None else existing["expenses"]
# If expenses came as pydantic models, convert to dicts
expenses_list = [e.model_dump() if hasattr(e, "model_dump") else e for e in expenses]
totals = self._calc_totals(salary, [ExpenseCategory(**e) for e in expenses_list])
update_doc = {
"$set": {
"salary": float(salary),
"expenses": expenses_list,
"total_expenses": totals["total_expenses"],
"remaining": totals["remaining"],
"updated_at": datetime.now(timezone.utc),
}
}
await self.col.update_one({"month_key": month_key}, update_doc)
updated = await self.col.find_one({"month_key": month_key})
return self._serialize(updated)
async def delete(self, month_key: str) -> None:
result = await self.col.delete_one({"month_key": month_key})
if result.deleted_count == 0:
raise HTTPException(status_code=404, detail="No record found for this month")
|