"""Default data seeding — EUR currency, 19% and 7% tax rates.""" from __future__ import annotations import logging import uuid from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.currency import Currency from app.models.tax import TaxRate logger = logging.getLogger(__name__) async def seed_default_data(db: AsyncSession, tenant_id: uuid.UUID | None = None) -> None: """Seed default currency and tax rates for a tenant. Idempotent: checks if data already exists before inserting. If tenant_id is None, seeds for all existing tenants. """ from app.models.tenant import Tenant if tenant_id is not None: tenant_ids = [tenant_id] else: result = await db.execute(select(Tenant.id)) tenant_ids = [row[0] for row in result.all()] for tid in tenant_ids: await _seed_currencies(db, tid) await _seed_tax_rates(db, tid) await db.flush() logger.info(f"Seeded default data for {len(tenant_ids)} tenant(s)") async def _seed_currencies(db: AsyncSession, tenant_id: uuid.UUID) -> None: """Seed EUR as default currency if no currencies exist for this tenant.""" existing = await db.execute( select(Currency).where( Currency.tenant_id == tenant_id, Currency.deleted_at.is_(None), ) ) if existing.scalars().first() is not None: return eur = Currency( tenant_id=tenant_id, code="EUR", name="Euro", symbol="€", is_default=True, ) db.add(eur) logger.info(f"Seeded EUR currency for tenant {tenant_id}") async def _seed_tax_rates(db: AsyncSession, tenant_id: uuid.UUID) -> None: """Seed 19% and 7% tax rates if no tax rates exist for this tenant.""" existing = await db.execute( select(TaxRate).where( TaxRate.tenant_id == tenant_id, TaxRate.deleted_at.is_(None), ) ) if existing.scalars().first() is not None: return vat_19 = TaxRate( tenant_id=tenant_id, name="Mehrwertsteuer 19%", rate=19.00, is_default=True, ) vat_7 = TaxRate( tenant_id=tenant_id, name="Mehrwertsteuer 7%", rate=7.00, is_default=False, ) db.add(vat_19) db.add(vat_7) logger.info(f"Seeded 19% and 7% tax rates for tenant {tenant_id}")