From a7fdaf21dc7b003bc564f8c837bc6e91472f3e4c Mon Sep 17 00:00:00 2001 From: Leopoldadmin Date: Sat, 4 Jul 2026 00:25:38 +0000 Subject: [PATCH] feat(core): add system settings service --- app/services/system_settings_service.py | 144 ++++++++++++++++++++++++ 1 file changed, 144 insertions(+) create mode 100644 app/services/system_settings_service.py diff --git a/app/services/system_settings_service.py b/app/services/system_settings_service.py new file mode 100644 index 0000000..734edfe --- /dev/null +++ b/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)