feat(core): add tax rate service

app/services/tax_service.py

@@ -0,0 +1,167 @@
+"""Tax rate service — CRUD, default tax management, seeding."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.audit import log_audit
+from app.models.tax import TaxRate
+
+
+def _tax_to_dict(t: TaxRate) -> dict[str, Any]:
+    """Serialize a TaxRate ORM object to dict."""
+    return {
+        "id": str(t.id),
+        "name": t.name,
+        "rate": float(t.rate),
+        "is_default": t.is_default,
+        "country": t.country,
+        "created_at": t.created_at.isoformat() if t.created_at else None,
+        "updated_at": t.updated_at.isoformat() if t.updated_at else None,
+    }
+
+
+async def list_tax_rates(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+) -> dict[str, Any]:
+    """List all tax rates for a tenant."""
+    q = select(TaxRate).where(
+        TaxRate.tenant_id == tenant_id,
+        TaxRate.deleted_at.is_(None),
+    ).order_by(TaxRate.rate.asc())
+    result = await db.execute(q)
+    tax_rates = result.scalars().all()
+    return {
+        "items": [_tax_to_dict(t) for t in tax_rates],
+        "total": len(tax_rates),
+    }
+
+
+async def create_tax_rate(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any]:
+    """Create a new tax rate. If is_default=True, unset other defaults first."""
+    if data.get("is_default"):
+        await db.execute(
+            update(TaxRate)
+            .where(TaxRate.tenant_id == tenant_id, TaxRate.is_default.is_(True))
+            .values(is_default=False)
+        )
+
+    tax_rate = TaxRate(
+        tenant_id=tenant_id,
+        name=data["name"],
+        rate=data["rate"],
+        is_default=data.get("is_default", False),
+        country=data.get("country"),
+    )
+    db.add(tax_rate)
+    await db.flush()
+    await db.refresh(tax_rate)
+    await log_audit(
+        db, tenant_id, user_id, "create", "tax_rate", tax_rate.id,
+        changes={"name": tax_rate.name, "rate": float(tax_rate.rate)},
+    )
+    return _tax_to_dict(tax_rate)
+
+
+async def update_tax_rate(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    tax_id: uuid.UUID,
+    data: dict[str, Any],
+) -> dict[str, Any] | None:
+    """Update a tax rate. If setting is_default=True, unset other defaults first."""
+    q = select(TaxRate).where(
+        TaxRate.id == tax_id,
+        TaxRate.tenant_id == tenant_id,
+        TaxRate.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    tax_rate = result.scalar_one_or_none()
+    if tax_rate is None:
+        return None
+
+    if data.get("is_default") is True and not tax_rate.is_default:
+        await db.execute(
+            update(TaxRate)
+            .where(
+                TaxRate.tenant_id == tenant_id,
+                TaxRate.is_default.is_(True),
+                TaxRate.id != tax_id,
+            )
+            .values(is_default=False)
+        )
+
+    changes: dict[str, Any] = {}
+    for field in ("name", "rate", "is_default", "country"):
+        if field in data and data[field] is not None:
+            old_val = getattr(tax_rate, field)
+            changes[field] = {"old": old_val, "new": data[field]}
+            setattr(tax_rate, field, data[field])
+
+    await db.flush()
+    await db.refresh(tax_rate)
+    await log_audit(db, tenant_id, user_id, "update", "tax_rate", tax_id, changes=changes)
+    return _tax_to_dict(tax_rate)
+
+
+async def delete_tax_rate(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    tax_id: uuid.UUID,
+) -> bool:
+    """Soft-delete a tax rate."""
+    q = select(TaxRate).where(
+        TaxRate.id == tax_id,
+        TaxRate.tenant_id == tenant_id,
+        TaxRate.deleted_at.is_(None),
+    )
+    result = await db.execute(q)
+    tax_rate = result.scalar_one_or_none()
+    if tax_rate is None:
+        return False
+
+    tax_rate.deleted_at = datetime.now(UTC)
+    await db.flush()
+    await log_audit(
+        db, tenant_id, user_id, "delete", "tax_rate", tax_id,
+        changes={"name": tax_rate.name},
+    )
+    return True
+
+
+async def seed_default_tax_rates(db: AsyncSession, tenant_id: uuid.UUID) -> None:
+    """Seed default 19% and 7% tax rates for a new tenant if none exist."""
+    q = select(TaxRate).where(TaxRate.tenant_id == tenant_id, TaxRate.deleted_at.is_(None))
+    result = await db.execute(q)
+    if result.scalars().first() is not None:
+        return
+
+    tax_19 = TaxRate(
+        tenant_id=tenant_id,
+        name="Mehrwertsteuer 19%",
+        rate=19.00,
+        is_default=True,
+        country="DE",
+    )
+    tax_7 = TaxRate(
+        tenant_id=tenant_id,
+        name="Mehrwertsteuer 7%",
+        rate=7.00,
+        is_default=False,
+        country="DE",
+    )
+    db.add_all([tax_19, tax_7])
+    await db.flush()