145 lines
5.6 KiB
Python
145 lines
5.6 KiB
Python
"""System settings service — get, upsert (singleton per tenant)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.audit import log_audit
|
|
from app.models.system_settings import SystemSettings
|
|
|
|
|
|
def _settings_to_dict(s: SystemSettings) -> dict[str, Any]:
|
|
"""Serialize SystemSettings ORM object to dict."""
|
|
return {
|
|
"id": str(s.id),
|
|
"company_name": s.company_name,
|
|
"company_legal_form": s.company_legal_form,
|
|
"company_street": s.company_street,
|
|
"company_city": s.company_city,
|
|
"company_zip": s.company_zip,
|
|
"company_country": s.company_country,
|
|
"tax_number": s.tax_number,
|
|
"vat_id": s.vat_id,
|
|
"iban": s.iban,
|
|
"bic": s.bic,
|
|
"bank_name": s.bank_name,
|
|
"ceo": s.ceo,
|
|
"trade_register": s.trade_register,
|
|
"default_currency_id": str(s.default_currency_id) if s.default_currency_id else None,
|
|
"default_tax_id": str(s.default_tax_id) if s.default_tax_id else None,
|
|
"invoice_prefix": s.invoice_prefix,
|
|
"quote_prefix": s.quote_prefix,
|
|
"payment_terms_days": s.payment_terms_days,
|
|
"created_at": s.created_at.isoformat() if s.created_at else None,
|
|
"updated_at": s.updated_at.isoformat() if s.updated_at else None,
|
|
}
|
|
|
|
|
|
async def get_system_settings(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
) -> dict[str, Any] | None:
|
|
"""Get the singleton system settings for a tenant."""
|
|
q = select(SystemSettings).where(
|
|
SystemSettings.tenant_id == tenant_id,
|
|
SystemSettings.deleted_at.is_(None),
|
|
)
|
|
result = await db.execute(q)
|
|
settings = result.scalar_one_or_none()
|
|
if settings is None:
|
|
return None
|
|
return _settings_to_dict(settings)
|
|
|
|
|
|
async def upsert_system_settings(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
data: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""Upsert system settings for a tenant. Creates if not exists, updates if exists."""
|
|
q = select(SystemSettings).where(
|
|
SystemSettings.tenant_id == tenant_id,
|
|
SystemSettings.deleted_at.is_(None),
|
|
)
|
|
result = await db.execute(q)
|
|
settings = result.scalar_one_or_none()
|
|
|
|
# Parse UUID fields
|
|
default_currency_id = data.get("default_currency_id")
|
|
if default_currency_id and isinstance(default_currency_id, str):
|
|
try:
|
|
default_currency_id = uuid.UUID(default_currency_id)
|
|
except ValueError:
|
|
default_currency_id = None
|
|
|
|
default_tax_id = data.get("default_tax_id")
|
|
if default_tax_id and isinstance(default_tax_id, str):
|
|
try:
|
|
default_tax_id = uuid.UUID(default_tax_id)
|
|
except ValueError:
|
|
default_tax_id = None
|
|
|
|
if settings is None:
|
|
# Create new settings
|
|
settings = SystemSettings(
|
|
tenant_id=tenant_id,
|
|
company_name=data["company_name"],
|
|
company_legal_form=data.get("company_legal_form"),
|
|
company_street=data["company_street"],
|
|
company_city=data["company_city"],
|
|
company_zip=data["company_zip"],
|
|
company_country=data["company_country"],
|
|
tax_number=data.get("tax_number"),
|
|
vat_id=data.get("vat_id"),
|
|
iban=data.get("iban"),
|
|
bic=data.get("bic"),
|
|
bank_name=data.get("bank_name"),
|
|
ceo=data.get("ceo"),
|
|
trade_register=data.get("trade_register"),
|
|
default_currency_id=default_currency_id,
|
|
default_tax_id=default_tax_id,
|
|
invoice_prefix=data.get("invoice_prefix", "RE-"),
|
|
quote_prefix=data.get("quote_prefix", "AN-"),
|
|
payment_terms_days=data.get("payment_terms_days", 14),
|
|
)
|
|
db.add(settings)
|
|
await db.flush()
|
|
await db.refresh(settings)
|
|
await log_audit(
|
|
db, tenant_id, user_id, "create", "system_settings", settings.id,
|
|
changes={"company_name": settings.company_name},
|
|
)
|
|
else:
|
|
# Update existing settings
|
|
changes: dict[str, Any] = {}
|
|
all_fields = (
|
|
"company_name", "company_legal_form", "company_street", "company_city",
|
|
"company_zip", "company_country", "tax_number", "vat_id", "iban",
|
|
"bic", "bank_name", "ceo", "trade_register",
|
|
"invoice_prefix", "quote_prefix", "payment_terms_days",
|
|
)
|
|
for field in all_fields:
|
|
if field in data and data[field] is not None:
|
|
old_val = getattr(settings, field)
|
|
changes[field] = {"old": old_val, "new": data[field]}
|
|
setattr(settings, field, data[field])
|
|
|
|
if default_currency_id is not None or data.get("default_currency_id") is None and "default_currency_id" in data:
|
|
changes["default_currency_id"] = {"old": str(settings.default_currency_id) if settings.default_currency_id else None, "new": str(default_currency_id) if default_currency_id else None}
|
|
settings.default_currency_id = default_currency_id
|
|
|
|
if default_tax_id is not None or data.get("default_tax_id") is None and "default_tax_id" in data:
|
|
changes["default_tax_id"] = {"old": str(settings.default_tax_id) if settings.default_tax_id else None, "new": str(default_tax_id) if default_tax_id else None}
|
|
settings.default_tax_id = default_tax_id
|
|
|
|
await db.flush()
|
|
await db.refresh(settings)
|
|
await log_audit(db, tenant_id, user_id, "update", "system_settings", settings.id, changes=changes)
|
|
|
|
return _settings_to_dict(settings)
|