from typing import List, Optional from sqlalchemy.orm import Session from app_core.db.database import SessionLocal from app_core.db.models import TriumphDebtorMapping from datetime import datetime class MappingsService: def __init__(self): pass def get_all_mappings(self) -> List[TriumphDebtorMapping]: with SessionLocal() as db: return db.query(TriumphDebtorMapping).order_by(TriumphDebtorMapping.id.asc()).all() def get_mapping_by_id(self, mapping_id: int) -> Optional[TriumphDebtorMapping]: with SessionLocal() as db: return db.query(TriumphDebtorMapping).filter(TriumphDebtorMapping.id == mapping_id).first() def create_mapping(self, code: str, name: str, dbmacc: str, outlet: str) -> TriumphDebtorMapping: with SessionLocal() as db: mapping = TriumphDebtorMapping( code=code, name=name, dbmacc=dbmacc, outlet=outlet ) db.add(mapping) db.commit() db.refresh(mapping) return mapping def update_mapping(self, mapping_id: int, code: str, name: str, dbmacc: str, outlet: str) -> bool: with SessionLocal() as db: mapping = db.query(TriumphDebtorMapping).filter(TriumphDebtorMapping.id == mapping_id).first() if mapping: mapping.code = code mapping.name = name mapping.dbmacc = dbmacc mapping.outlet = outlet mapping.updated_at = datetime.now() db.commit() return True return False def delete_mapping(self, mapping_id: int) -> bool: with SessionLocal() as db: mapping = db.query(TriumphDebtorMapping).filter(TriumphDebtorMapping.id == mapping_id).first() if mapping: db.delete(mapping) db.commit() return True return False