diff --git a/app/services/tax_service.py b/app/services/tax_service.py new file mode 100644 index 0000000..696a7df --- /dev/null +++ b/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()