53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
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
|