""" HSEQ SaaS Multi-Tenant — Company Manager CRUD operaties voor tenant-beheer (Company model). Alle operaties zijn tenant-isolatie veilig: RLS wordt gehandhaafd op database-niveau [1]. Bronnen: [1] PostgreSQL RLS — postgresql.org/docs/current/ddl-rowsecurity.html [2] Phase 2 Database Architectuur — phase2_database_architecture_v1.0.md """ from __future__ import annotations import re import uuid from datetime import datetime, timezone from typing import Optional from sqlalchemy import text from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from models_v1_0 import ( VALID_BRZO_TIERS, VALID_PLANS, VALID_STATUSES, Company, CompanyModule, PLAN_DEFAULT_MODULES, ) # --------------------------------------------------------------------------- # Slug generatie # --------------------------------------------------------------------------- _SLUG_PATTERN = re.compile(r"[^a-z0-9\-]") def generate_slug(name: str) -> str: """ Genereert een URL-safe slug uit een bedrijfsnaam. Voorbeeld: 'Acme Chemie B.V.' → 'acme-chemie-bv' Args: name: Bedrijfsnaam. Returns: URL-safe slug string. """ slug = name.lower().strip() slug = _SLUG_PATTERN.sub("-", slug) slug = re.sub(r"-{2,}", "-", slug) slug = slug.strip("-") return slug[:100] def _unique_slug(session: Session, base_slug: str) -> str: """ Garandeert een unieke slug door een numeriek suffix toe te voegen bij conflict. Args: session: Actieve SQLAlchemy session. base_slug: Gewenste slug. Returns: Unieke slug. """ candidate = base_slug suffix = 1 while session.query(Company).filter_by(slug=candidate).first() is not None: candidate = f"{base_slug}-{suffix}" suffix += 1 return candidate # --------------------------------------------------------------------------- # Validatie helpers # --------------------------------------------------------------------------- def _validate_plan(plan: str) -> None: if plan not in VALID_PLANS: raise ValueError( f"Ongeldig plan '{plan}'. Geldige waarden: {', '.join(VALID_PLANS)}" ) def _validate_status(status: str) -> None: if status not in VALID_STATUSES: raise ValueError( f"Ongeldige status '{status}'. Geldige waarden: {', '.join(VALID_STATUSES)}" ) def _validate_brzo_tier(tier: Optional[str]) -> None: if tier is not None and tier not in VALID_BRZO_TIERS: raise ValueError( f"Ongeldige BRZO tier '{tier}'. Geldige waarden: {', '.join(VALID_BRZO_TIERS)}" ) # --------------------------------------------------------------------------- # CRUD operaties # --------------------------------------------------------------------------- def create_company( session: Session, *, name: str, contact_email: str, plan: str = "starter", legal_name: Optional[str] = None, kvk_number: Optional[str] = None, brzo_tier: Optional[str] = None, address: Optional[dict] = None, contact_phone: Optional[str] = None, logo_url: Optional[str] = None, settings: Optional[dict] = None, ) -> Company: """ Maakt een nieuw bedrijf (tenant) aan met bijbehorende standaardmodules. Bij aanmaak worden automatisch de modules ingeschakeld die bij het geselecteerde abonnement (plan) horen, conform PLAN_DEFAULT_MODULES [2]. Args: session: Actieve SQLAlchemy session. name: Bedrijfsnaam (verplicht). contact_email: Primair contact e-mailadres (verplicht). plan: Abonnementstype (default: starter). legal_name: Juridische naam (optioneel). kvk_number: Kamer van Koophandel nummer (optioneel). brzo_tier: Seveso-classificatie (optioneel). address: Adresgegevens als dict (optioneel). contact_phone: Telefoonnummer (optioneel). logo_url: URL naar bedrijfslogo (optioneel). settings: Tenant-specifieke configuratie als dict (optioneel). Returns: Aangemaakt Company-object met geïnitialiseerde modules. Raises: ValueError: Bij ongeldige invoerparameters. IntegrityError: Bij slug-conflict (zeldzaam na suffix-logging). """ _validate_plan(plan) _validate_brzo_tier(brzo_tier) slug = _unique_slug(session, generate_slug(name)) company = Company( slug=slug, name=name, legal_name=legal_name, kvk_number=kvk_number, brzo_tier=brzo_tier, address=address, contact_email=contact_email, contact_phone=contact_phone, logo_url=logo_url, plan=plan, status="active", settings=settings or {}, ) session.add(company) session.flush() # Zorg dat company.id beschikbaar is # Schakel standaardmodules in op basis van plan enabled_modules = PLAN_DEFAULT_MODULES.get(plan, PLAN_DEFAULT_MODULES["starter"]) for module_key in enabled_modules: company_module = CompanyModule( company_id=company.id, module_key=module_key, enabled=True, config={}, ) session.add(company_module) return company def get_company(session: Session, company_id: uuid.UUID) -> Optional[Company]: """ Haalt een bedrijf op basis van UUID. Args: session: Actieve SQLAlchemy session. company_id: UUID van het bedrijf. Returns: Company-object of None indien niet gevonden. """ return session.query(Company).filter_by(id=company_id).first() def get_company_by_slug(session: Session, slug: str) -> Optional[Company]: """ Haalt een bedrijf op basis van URL-safe slug. Args: session: Actieve SQLAlchemy session. slug: URL-safe identifier. Returns: Company-object of None indien niet gevonden. """ return session.query(Company).filter_by(slug=slug).first() def list_companies( session: Session, *, status: Optional[str] = None, plan: Optional[str] = None, limit: int = 50, offset: int = 0, ) -> list[Company]: """ Lijst van bedrijven met optionele filtering op status en plan. Args: session: Actieve SQLAlchemy session. status: Filter op status (active, suspended, churned, trial). plan: Filter op abonnementstype. limit: Maximum aantal resultaten (default: 50). offset: Offset voor paginering. Returns: Lijst van Company-objecten. """ query = session.query(Company) if status: _validate_status(status) query = query.filter_by(status=status) if plan: _validate_plan(plan) query = query.filter_by(plan=plan) return query.order_by(Company.created_at.desc()).offset(offset).limit(limit).all() def update_company( session: Session, company_id: uuid.UUID, *, name: Optional[str] = None, legal_name: Optional[str] = None, kvk_number: Optional[str] = None, brzo_tier: Optional[str] = None, address: Optional[dict] = None, contact_email: Optional[str] = None, contact_phone: Optional[str] = None, logo_url: Optional[str] = None, plan: Optional[str] = None, status: Optional[str] = None, settings: Optional[dict] = None, ) -> Optional[Company]: """ Werkt bedrijfsgegevens bij. Alleen opgegeven velden worden gewijzigd. Bij plan-wijziging worden niet-beschikbare modules automatisch uitgeschakeld. Args: session: Actieve SQLAlchemy session. company_id: UUID van het bedrijf. name: Nieuwe bedrijfsnaam (optioneel). plan: Nieuw abonnementstype (optioneel). status: Nieuwe status (optioneel). ... overige optionele velden. Returns: Geüpdatet Company-object of None indien niet gevonden. Raises: ValueError: Bij ongeldige invoerparameters. """ company = get_company(session, company_id) if company is None: return None if plan is not None: _validate_plan(plan) company.plan = plan # Pas module-beschikbaarheid aan op nieuw plan _sync_modules_to_plan(session, company) if status is not None: _validate_status(status) company.status = status if brzo_tier is not None: _validate_brzo_tier(brzo_tier) company.brzo_tier = brzo_tier if name is not None: company.name = name if legal_name is not None: company.legal_name = legal_name if kvk_number is not None: company.kvk_number = kvk_number if address is not None: company.address = address if contact_email is not None: company.contact_email = contact_email if contact_phone is not None: company.contact_phone = contact_phone if logo_url is not None: company.logo_url = logo_url if settings is not None: company.settings = settings company.updated_at = datetime.now(timezone.utc) return company def delete_company(session: Session, company_id: uuid.UUID) -> bool: """ Deactiveert een bedrijf (soft delete via status = 'churned'). Hard delete wordt bewust NIET ondersteund i.v.m. audit-trail en BRZO-bewaarplicht (minimaal 5 jaar) [2]. Args: session: Actieve SQLAlchemy session. company_id: UUID van het bedrijf. Returns: True indien succesvol gedeactiveerd, False indien niet gevonden. """ company = get_company(session, company_id) if company is None: return False company.status = "churned" company.updated_at = datetime.now(timezone.utc) return True def _sync_modules_to_plan(session: Session, company: Company) -> None: """ Synchroniseert ingeschakelde modules met het abonnement. Modules buiten het plan worden uitgeschakeld, modules binnen het plan die nog niet bestaan worden aangemaakt. Args: session: Actieve SQLAlchemy session. company: Company-object met bijgewerkt plan. """ allowed = set(PLAN_DEFAULT_MODULES.get(company.plan, [])) existing_modules = ( session.query(CompanyModule) .filter_by(company_id=company.id) .all() ) existing_keys = {m.module_key for m in existing_modules} # Schakel modules uit die niet in het nieuwe plan vallen for module in existing_modules: if module.module_key not in allowed: module.enabled = False # Voeg ontbrekende modules toe for module_key in allowed - existing_keys: session.add(CompanyModule( company_id=company.id, module_key=module_key, enabled=True, config={}, ))