feat(core): add system settings service

app/services/system_settings_service.py

@@ -0,0 +1,144 @@
+"""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)