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.value ) 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.value, settings=settings_data.settings ) db.add(new_settings) db.commit() db.refresh(new_settings) return new_settings