""" HSEQ SaaS Multi-Tenant — SQLAlchemy ORM Models Definieert alle database-modellen voor multi-tenant beheer: - Company (tenant registry) - CompanyUser (user-company mapping met RBAC) - CompanyModule (module toggles per tenant) - CompanyPermission (fine-grained RBAC per tenant) - ModuleCatalog (systeem-brede module-definities) Architectuur: PostgreSQL met Row-Level Security (RLS) [1][2]. Shared database, shared schema, company_id FK als tenant-isolatie [3]. Bronnen: [1] PostgreSQL RLS — postgresql.org/docs/current/ddl-rowsecurity.html [2] Phase 2 Database Architectuur — phase2_database_architecture_v1.0.md [3] Multi-tenant SaaS patterns — AWS Well-Architected Framework """ from __future__ import annotations import uuid from datetime import datetime, timezone from typing import Optional from sqlalchemy import ( Boolean, Column, ForeignKey, Index, String, Text, UniqueConstraint, event, text, ) from sqlalchemy.dialects.postgresql import JSONB, UUID from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship # --------------------------------------------------------------------------- # Base # --------------------------------------------------------------------------- class Base(DeclarativeBase): """Declarative base voor alle modellen.""" pass class TenantMixin: """ Mixin voor tenant-scoped modellen. Voegt company_id toe met automatische populatie via session events. Gebruik: class Incident(TenantMixin, Base): __tablename__ = 'incidents' ... """ company_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("companies.id", ondelete="CASCADE"), nullable=False, index=True, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _utcnow() -> datetime: return datetime.now(timezone.utc) def _new_uuid() -> uuid.UUID: return uuid.uuid4() # --------------------------------------------------------------------------- # Company (Tenant Registry) # --------------------------------------------------------------------------- class Company(Base): """ Centrale tenant-registratie. Elke rij vertegenwoordigt één bedrijf. UUID primary key voorkomt orde-lek tussen tenants bij auto-increment [2]. `slug` wordt gebruikt voor URL-routing en API-namespacing. `settings` als JSONB voor flexibele per-tenant configuratie. """ __tablename__ = "companies" id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=_new_uuid ) slug: Mapped[str] = mapped_column( String(100), unique=True, nullable=False, index=True, comment="URL-safe identifier, bijv. 'acme-chemie'", ) name: Mapped[str] = mapped_column(String(200), nullable=False) legal_name: Mapped[Optional[str]] = mapped_column(String(200), nullable=True) kvk_number: Mapped[Optional[str]] = mapped_column( String(20), nullable=True, comment="Kamer van Koophandel nummer", ) brzo_tier: Mapped[Optional[str]] = mapped_column( String(20), nullable=True, comment="Seveso-classificatie: 'lower', 'upper', of 'none'", ) address: Mapped[Optional[dict]] = mapped_column( JSONB, nullable=True, comment="Adresgegevens: {street, city, postal_code, country}", ) contact_email: Mapped[str] = mapped_column(String(255), nullable=False) contact_phone: Mapped[Optional[str]] = mapped_column(String(50), nullable=True) logo_url: Mapped[Optional[str]] = mapped_column(Text, nullable=True) plan: Mapped[str] = mapped_column( String(50), nullable=False, default="starter", comment="Abonnement: starter, professional, enterprise", ) status: Mapped[str] = mapped_column( String(20), nullable=False, default="active", index=True, comment="Status: active, suspended, churned, trial", ) settings: Mapped[dict] = mapped_column( JSONB, nullable=False, default=dict, comment="Tenant-specifieke configuratie (branding, features, etc.)", ) created_at: Mapped[datetime] = mapped_column( default=_utcnow, nullable=False, ) updated_at: Mapped[datetime] = mapped_column( default=_utcnow, onupdate=_utcnow, nullable=False, ) # --- Relationships --- users: Mapped[list["CompanyUser"]] = relationship( "CompanyUser", back_populates="company", lazy="dynamic", ) modules: Mapped[list["CompanyModule"]] = relationship( "CompanyModule", back_populates="company", lazy="dynamic", ) # --- Constraints --- __table_args__ = ( Index("idx_companies_slug", "slug"), Index("idx_companies_status", "status"), Index("idx_companies_plan", "plan"), ) def __repr__(self) -> str: return f"" # --------------------------------------------------------------------------- # CompanyUser — User-Company Mapping # --------------------------------------------------------------------------- class CompanyUser(Base): """ Koppelt gebruikers aan bedrijven met rol-toewijzing. Ondersteunt multi-company gebruikers (bijv. auditors, consultants). UNIQUE (user_id, company_id) voorkomt dubbele koppelingen. """ __tablename__ = "company_users" id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=_new_uuid ) user_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True, ) company_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("companies.id", ondelete="CASCADE"), nullable=False, index=True, ) role: Mapped[str] = mapped_column( String(50), nullable=False, comment="Rol: owner, admin, manager, user, viewer", ) status: Mapped[str] = mapped_column( String(20), nullable=False, default="active", comment="Status: active, invited, deactivated", ) invited_at: Mapped[Optional[datetime]] = mapped_column(nullable=True) joined_at: Mapped[datetime] = mapped_column(default=_utcnow, nullable=False) # --- Relationships --- company: Mapped["Company"] = relationship("Company", back_populates="users") # --- Constraints --- __table_args__ = ( UniqueConstraint("user_id", "company_id", name="uq_company_users_user_company"), Index("idx_company_users_company", "company_id"), Index("idx_company_users_user", "user_id"), Index("idx_company_users_role", "company_id", "role"), ) def __repr__(self) -> str: return ( f"" ) # --------------------------------------------------------------------------- # CompanyModule — Module Toggles per Tenant # --------------------------------------------------------------------------- class CompanyModule(Base): """ Bepaalt welke HSEQ-modules beschikbaar zijn per tenant. Standaard ingeschakelde modules zijn afhankelijk van het abonnement (plan). Module keys: compliance, incidents, risk_assessment, ptw, moc, training, environment, brzo, intelligence, agents, knowledge [2]. """ __tablename__ = "company_modules" id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=_new_uuid ) company_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("companies.id", ondelete="CASCADE"), nullable=False, index=True, ) module_key: Mapped[str] = mapped_column( String(50), nullable=False, comment="Module identifier, bijv. 'compliance', 'incidents', 'ptw'", ) enabled: Mapped[bool] = mapped_column( Boolean, nullable=False, default=True, ) config: Mapped[Optional[dict]] = mapped_column( JSONB, default=dict, comment="Module-specifieke configuratie", ) enabled_at: Mapped[Optional[datetime]] = mapped_column( default=_utcnow, nullable=True, ) # --- Relationships --- company: Mapped["Company"] = relationship("Company", back_populates="modules") # --- Constraints --- __table_args__ = ( UniqueConstraint("company_id", "module_key", name="uq_company_modules_company_key"), Index("idx_company_modules_company", "company_id"), Index("idx_company_modules_enabled", "company_id", "enabled"), ) def __repr__(self) -> str: state = "ON" if self.enabled else "OFF" return f"" # --------------------------------------------------------------------------- # CompanyPermission — RBAC per Tenant # --------------------------------------------------------------------------- class CompanyPermission(Base): """ Rol-gebaseerd toegangsbeheer per tenant. Ondersteunt coarse-grained (module-level) en fine-grained (actie-level) permissies. Default permissie-matrix per rol [2]: owner/admin → CRUD+X op alle modules manager → CRU op compliance/incidents/moc/rie, CRUD op ptw, R op rest user → CR op compliance/incidents/ptw/moc, R op rest viewer → R op alles """ __tablename__ = "company_permissions" id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=_new_uuid ) role: Mapped[str] = mapped_column( String(50), nullable=False, comment="Rol: owner, admin, manager, user, viewer", ) resource: Mapped[str] = mapped_column( String(100), nullable=False, comment="Module/resource naam: 'incidents', 'ptw', 'compliance'", ) action: Mapped[str] = mapped_column( String(50), nullable=False, comment="Actie: create, read, update, delete, export, approve", ) allowed: Mapped[bool] = mapped_column( Boolean, nullable=False, default=True, ) company_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("companies.id", ondelete="CASCADE"), nullable=True, comment="NULL = globale default, UUID = tenant-specifieke override", ) # --- Constraints --- __table_args__ = ( UniqueConstraint( "role", "resource", "action", "company_id", name="uq_company_permissions_role_resource_action", ), Index("idx_company_permissions_lookup", "role", "resource", "company_id"), ) def __repr__(self) -> str: state = "ALLOW" if self.allowed else "DENY" return ( f"" ) # --------------------------------------------------------------------------- # ModuleCatalog — Systeem-brede module-definities # --------------------------------------------------------------------------- class ModuleCatalog(Base): """ Module-definities op platform-niveau. Niet per tenant — dit is de bron van beschikbare modules. Categorieën: core (starter+), advanced (professional+), premium (enterprise). """ __tablename__ = "module_catalog" id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=_new_uuid ) code: Mapped[str] = mapped_column( String(50), unique=True, nullable=False, comment="Unieke module-code, bijv. 'compliance', 'intelligence'", ) name: Mapped[str] = mapped_column(String(100), nullable=False) description: Mapped[Optional[str]] = mapped_column(Text, nullable=True) category: Mapped[str] = mapped_column( String(50), nullable=False, comment="Categorie: core, advanced, premium", ) version: Mapped[str] = mapped_column(String(20), default="1.0", nullable=False) created_at: Mapped[datetime] = mapped_column(default=_utcnow, nullable=False) def __repr__(self) -> str: return f"" # --------------------------------------------------------------------------- # Default module seed data (gebruikt bij company-aanmaak) # --------------------------------------------------------------------------- PLAN_DEFAULT_MODULES: dict[str, list[str]] = { "starter": [ "compliance", "incidents", "risk_assessment", "ptw", "moc", "training", ], "professional": [ "compliance", "incidents", "risk_assessment", "ptw", "moc", "training", "environment", "brzo", "intelligence", ], "enterprise": [ "compliance", "incidents", "risk_assessment", "ptw", "moc", "training", "environment", "brzo", "intelligence", "agents", "knowledge", ], } VALID_ROLES: tuple[str, ...] = ("owner", "admin", "manager", "user", "viewer") VALID_PLANS: tuple[str, ...] = ("starter", "professional", "enterprise") VALID_STATUSES: tuple[str, ...] = ("active", "suspended", "churned", "trial") VALID_BRZO_TIERS: tuple[str, ...] = ("lower", "upper", "none") VALID_PERMISSION_ACTIONS: tuple[str, ...] = ( "create", "read", "update", "delete", "export", "approve", )