Spaces:
Paused
Paused
| """ | |
| Transaction Service - Business logic for transaction management | |
| """ | |
| import logging | |
| from typing import Any, Optional | |
| from sqlalchemy.orm import Session | |
| from core.database import Transaction | |
| from .repository import TransactionRepository | |
| logger = logging.getLogger(__name__) | |
| class TransactionService: | |
| """Service for managing financial transactions""" | |
| def get_transactions_by_case( | |
| self, db: Session, case_id: str, filters: dict[str, Any] | None = None | |
| ) -> list[Transaction]: | |
| """Get all transactions for a case""" | |
| repo = TransactionRepository(db) | |
| return repo.get_by_case_id(case_id, filters) | |
| def get_transaction( | |
| self, db: Session, transaction_id: str | |
| ) -> Optional[Transaction]: | |
| """Get a single transaction by ID""" | |
| repo = TransactionRepository(db) | |
| return repo.get_by_id(transaction_id) | |
| def create_transaction(self, db: Session, transaction_data: Any) -> Transaction: | |
| """Create a new transaction""" | |
| repo = TransactionRepository(db) | |
| # Convert Pydantic model to dict if needed | |
| if hasattr(transaction_data, "dict"): | |
| data = transaction_data.dict() | |
| elif hasattr(transaction_data, "model_dump"): | |
| data = transaction_data.model_dump() | |
| else: | |
| data = transaction_data | |
| transaction = repo.create(data) | |
| db.commit() | |
| db.refresh(transaction) | |
| return transaction | |
| def update_transaction_status( | |
| self, db: Session, transaction_id: str, status: str, notes: str | None = None | |
| ) -> Optional[Transaction]: | |
| """Update transaction status""" | |
| repo = TransactionRepository(db) | |
| transaction = repo.get_by_id(transaction_id) | |
| if not transaction: | |
| return None | |
| updates = {"status": status} | |
| if notes: | |
| updates["notes"] = notes | |
| updated_transaction = repo.update(transaction, updates) | |
| db.commit() | |
| db.refresh(updated_transaction) | |
| return updated_transaction | |
| def delete_transaction(self, db: Session, transaction_id: str) -> bool: | |
| """Delete a transaction""" | |
| repo = TransactionRepository(db) | |
| transaction = repo.get_by_id(transaction_id) | |
| if not transaction: | |
| return False | |
| repo.delete(transaction) | |
| db.commit() | |
| return True | |
| def get_transaction_aggregates( | |
| self, | |
| db: Session, | |
| case_id: str | None = None, | |
| filters: dict[str, Any] | None = None, | |
| ) -> dict[str, Any]: | |
| """Get transaction aggregates (total, average, etc.)""" | |
| repo = TransactionRepository(db) | |
| return repo.get_aggregates(case_id, filters) | |
| # Singleton instance | |
| transaction_service = TransactionService() | |