71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import select
|
|
from fastapi import HTTPException
|
|
from api.v1.modules.a76.invoice_settings.models import InvoiceSettings
|
|
from api.v1.modules.a76.invoice_settings.dto import InvoiceSettingsRequest, OperationType
|
|
|
|
def get_settings(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
invoice_type: str,
|
|
operation_type: OperationType
|
|
) -> Optional[InvoiceSettings]:
|
|
"""Retrieve settings for a specific context"""
|
|
stmt = select(InvoiceSettings).where(
|
|
InvoiceSettings.tenant_id == tenant_id,
|
|
InvoiceSettings.company_id == company_id,
|
|
InvoiceSettings.invoice_type == invoice_type,
|
|
InvoiceSettings.operation_type == operation_type
|
|
)
|
|
return db.execute(stmt).scalar_one_or_none()
|
|
|
|
def list_settings(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int
|
|
) -> List[InvoiceSettings]:
|
|
"""List all settings for a company"""
|
|
stmt = select(InvoiceSettings).where(
|
|
InvoiceSettings.tenant_id == tenant_id,
|
|
InvoiceSettings.company_id == company_id
|
|
)
|
|
return db.execute(stmt).scalars().all()
|
|
|
|
def upsert_settings(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
settings_data: InvoiceSettingsRequest
|
|
) -> InvoiceSettings:
|
|
"""Create or update settings"""
|
|
# Check if exists
|
|
existing = get_settings(
|
|
db,
|
|
tenant_id,
|
|
company_id,
|
|
settings_data.invoice_type,
|
|
settings_data.operation_type
|
|
)
|
|
|
|
if existing:
|
|
existing.settings = settings_data.settings
|
|
db.commit()
|
|
db.refresh(existing)
|
|
return existing
|
|
|
|
# Create new
|
|
new_settings = InvoiceSettings(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
invoice_type=settings_data.invoice_type,
|
|
operation_type=settings_data.operation_type,
|
|
settings=settings_data.settings
|
|
)
|
|
|
|
db.add(new_settings)
|
|
db.commit()
|
|
db.refresh(new_settings)
|
|
return new_settings
|