138 lines
4.5 KiB
Python
138 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import and_, func
|
|
|
|
from adapters.db.models.petty_cash import PettyCash
|
|
from adapters.db.repositories.petty_cash_repository import PettyCashRepository
|
|
from app.core.responses import ApiError
|
|
|
|
|
|
def create_petty_cash(db: Session, business_id: int, data: Dict[str, Any]) -> Dict[str, Any]:
|
|
# validate required fields
|
|
name = (data.get("name") or "").strip()
|
|
if name == "":
|
|
raise ApiError("STRING_TOO_SHORT", "Name is required", http_status=400)
|
|
|
|
# code uniqueness in business if provided; else auto-generate numeric min 3 digits
|
|
code = data.get("code")
|
|
if code is not None and str(code).strip() != "":
|
|
if not str(code).isdigit():
|
|
raise ApiError("INVALID_PETTY_CASH_CODE", "Petty cash code must be numeric", http_status=400)
|
|
if len(str(code)) < 3:
|
|
raise ApiError("INVALID_PETTY_CASH_CODE", "Petty cash code must be at least 3 digits", http_status=400)
|
|
exists = db.query(PettyCash).filter(and_(PettyCash.business_id == business_id, PettyCash.code == str(code))).first()
|
|
if exists:
|
|
raise ApiError("DUPLICATE_PETTY_CASH_CODE", "Duplicate petty cash code", http_status=400)
|
|
else:
|
|
max_code = db.query(func.max(PettyCash.code)).filter(PettyCash.business_id == business_id).scalar()
|
|
try:
|
|
if max_code is not None and str(max_code).isdigit():
|
|
next_code_int = int(max_code) + 1
|
|
else:
|
|
next_code_int = 100
|
|
if next_code_int < 100:
|
|
next_code_int = 100
|
|
code = str(next_code_int)
|
|
except Exception:
|
|
code = "100"
|
|
|
|
repo = PettyCashRepository(db)
|
|
obj = repo.create(business_id, {
|
|
"name": name,
|
|
"code": code,
|
|
"description": data.get("description"),
|
|
"currency_id": int(data["currency_id"]),
|
|
"is_active": bool(data.get("is_active", True)),
|
|
"is_default": bool(data.get("is_default", False)),
|
|
})
|
|
|
|
# ensure single default
|
|
if obj.is_default:
|
|
repo.clear_default(business_id, except_id=obj.id)
|
|
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return petty_cash_to_dict(obj)
|
|
|
|
|
|
def get_petty_cash_by_id(db: Session, id_: int) -> Optional[Dict[str, Any]]:
|
|
obj = db.query(PettyCash).filter(PettyCash.id == id_).first()
|
|
return petty_cash_to_dict(obj) if obj else None
|
|
|
|
|
|
def update_petty_cash(db: Session, id_: int, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
repo = PettyCashRepository(db)
|
|
obj = repo.get_by_id(id_)
|
|
if obj is None:
|
|
return None
|
|
|
|
# validate name if provided
|
|
if "name" in data:
|
|
name_val = (data.get("name") or "").strip()
|
|
if name_val == "":
|
|
raise ApiError("STRING_TOO_SHORT", "Name is required", http_status=400)
|
|
|
|
if "code" in data and data["code"] is not None and str(data["code"]).strip() != "":
|
|
if not str(data["code"]).isdigit():
|
|
raise ApiError("INVALID_PETTY_CASH_CODE", "Petty cash code must be numeric", http_status=400)
|
|
if len(str(data["code"])) < 3:
|
|
raise ApiError("INVALID_PETTY_CASH_CODE", "Petty cash code must be at least 3 digits", http_status=400)
|
|
exists = db.query(PettyCash).filter(and_(PettyCash.business_id == obj.business_id, PettyCash.code == str(data["code"]), PettyCash.id != obj.id)).first()
|
|
if exists:
|
|
raise ApiError("DUPLICATE_PETTY_CASH_CODE", "Duplicate petty cash code", http_status=400)
|
|
|
|
repo.update(obj, data)
|
|
if obj.is_default:
|
|
repo.clear_default(obj.business_id, except_id=obj.id)
|
|
|
|
db.commit()
|
|
db.refresh(obj)
|
|
return petty_cash_to_dict(obj)
|
|
|
|
|
|
def delete_petty_cash(db: Session, id_: int) -> bool:
|
|
obj = db.query(PettyCash).filter(PettyCash.id == id_).first()
|
|
if obj is None:
|
|
return False
|
|
db.delete(obj)
|
|
db.commit()
|
|
return True
|
|
|
|
|
|
def bulk_delete_petty_cash(db: Session, business_id: int, ids: List[int]) -> Dict[str, Any]:
|
|
repo = PettyCashRepository(db)
|
|
result = repo.bulk_delete(business_id, ids)
|
|
try:
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise ApiError("BULK_DELETE_FAILED", "Bulk delete failed for petty cash", http_status=500)
|
|
return result
|
|
|
|
|
|
def list_petty_cash(db: Session, business_id: int, query: Dict[str, Any]) -> Dict[str, Any]:
|
|
repo = PettyCashRepository(db)
|
|
res = repo.list(business_id, query)
|
|
return {
|
|
"items": [petty_cash_to_dict(i) for i in res["items"]],
|
|
"pagination": res["pagination"],
|
|
"query_info": res["query_info"],
|
|
}
|
|
|
|
|
|
def petty_cash_to_dict(obj: PettyCash) -> Dict[str, Any]:
|
|
return {
|
|
"id": obj.id,
|
|
"business_id": obj.business_id,
|
|
"name": obj.name,
|
|
"code": obj.code,
|
|
"description": obj.description,
|
|
"currency_id": obj.currency_id,
|
|
"is_active": bool(obj.is_active),
|
|
"is_default": bool(obj.is_default),
|
|
"created_at": obj.created_at.isoformat(),
|
|
"updated_at": obj.updated_at.isoformat(),
|
|
}
|