"""Budget service""" from typing import Optional, List from datetime import datetime from sqlalchemy.orm import Session from app.db.repositories import BudgetRepository, TransactionRepository, CategoryRepository from app.db.models import Budget, TransactionType from app.schemas import BudgetCreateSchema class BudgetService: """Service for budget operations""" def __init__(self, session: Session): self.session = session self.budget_repo = BudgetRepository(session) self.transaction_repo = TransactionRepository(session) self.category_repo = CategoryRepository(session) def create_budget(self, family_id: int, data: BudgetCreateSchema) -> Budget: """Create new budget""" return self.budget_repo.create( family_id=family_id, name=data.name, limit_amount=data.limit_amount, period=data.period, alert_threshold=data.alert_threshold, category_id=data.category_id, start_date=data.start_date, ) def get_budget_status(self, budget_id: int) -> dict: """Get budget status with spent amount and percentage""" budget = self.budget_repo.get_by_id(budget_id) if not budget: return {} spent_percent = (budget.spent_amount / budget.limit_amount * 100) if budget.limit_amount > 0 else 0 remaining = budget.limit_amount - budget.spent_amount is_exceeded = spent_percent > 100 is_warning = spent_percent >= budget.alert_threshold return { "budget_id": budget.id, "name": budget.name, "limit": budget.limit_amount, "spent": budget.spent_amount, "remaining": remaining, "spent_percent": spent_percent, "is_exceeded": is_exceeded, "is_warning": is_warning, "alert_threshold": budget.alert_threshold, } def get_family_budget_status(self, family_id: int) -> List[dict]: """Get status of all budgets in family""" budgets = self.budget_repo.get_family_budgets(family_id) return [self.get_budget_status(budget.id) for budget in budgets] def check_budget_exceeded(self, budget_id: int) -> bool: """Check if budget limit exceeded""" status = self.get_budget_status(budget_id) return status.get("is_exceeded", False) def reset_budget(self, budget_id: int) -> Optional[Budget]: """Reset budget spent amount for new period""" return self.budget_repo.update(budget_id, spent_amount=0.0)