Domain-Driven Design con Multi-Tenancy in Python

  • Python
  • Gabriele Romanato

Progettare un sistema multi-tenant non significa solo “aggiungere un tenant_id ovunque”. Se vuoi mantenere il dominio espressivo, testabile e resistente al cambiamento, conviene affrontare la multi-tenancy come una preoccupazione architetturale che attraversa bounded context, persistenza, sicurezza e osservabilità, senza contaminare inutilmente la logica di business.

In questo articolo vediamo come applicare i principi di Domain-Driven Design (DDD) in un’applicazione Python multi-tenant, con esempi concreti: modellazione del dominio, isolamento dei dati (strategie e trade-off), infrastruttura (repository, unit of work, middleware), e controlli di sicurezza per evitare fughe di dati tra tenant.

1. Obiettivi e vincoli della multi-tenancy

Prima di modellare, chiarisci i requisiti. Le differenze tra progetti multi-tenant sono spesso qui:

  • Isolamento: quanto deve essere forte? (nessuna commistione a livello DB, o basta a livello logico?)
  • Personalizzazioni: ogni tenant ha configurazioni, regole, piani tariffari, feature flag?
  • Scalabilità: i tenant sono tanti e piccoli o pochi e grandi?
  • Compliance: requisiti di data residency, audit, crittografia, retention.
  • Operatività: migrazioni, backup/restore per tenant, onboarding/offboarding.

DDD non risolve automaticamente questi aspetti, ma ti aiuta a separare correttamente le responsabilità: il dominio resta pulito, mentre la multi-tenancy viene gestita da infrastruttura e politiche applicative.

2. Strategie di isolamento dati

In pratica, le strategie più comuni sono tre. Non esiste “la migliore”; esiste la migliore per i tuoi vincoli.

2.1 Database per tenant

  • Pro: isolamento massimo; backup/restore per tenant semplice; limiti di blast radius.
  • Contro: costo operativo; migrazioni moltiplicate; gestione connessioni.

2.2 Schema per tenant (PostgreSQL)

  • Pro: buon isolamento; una singola istanza DB; gestione accessi per schema.
  • Contro: migrazioni più complesse; tooling e ORM devono supportare bene.

2.3 Tabelle condivise con tenant_id

  • Pro: operazioni e migrazioni più semplici; costo più basso; onboarding rapido.
  • Contro: rischio di fuga dati se il filtro tenant è dimenticato; indici più complessi; alcune query diventano più pesanti.

DDD entra in gioco quando devi evitare che la scelta tecnica “invada” il modello. L’obiettivo è: le invarianti di dominio non devono dipendere dal fatto che i dati siano isolati per schema o per colonna.

3. Il tenant come concetto di dominio

Un errore tipico: trattare il tenant come un dettaglio infrastrutturale e poi scoprire che influenza regole di business (piani, limiti, pricing, permessi). Oppure il contrario: far “girare” il tenant ovunque nel dominio, creando rumore.

Una regola pratica:

  • Se il tenant influenza comportamenti di business (limiti, policy, configurazioni), allora esiste nel dominio come concetto (es. Organization, Account, Workspace).
  • Se serve solo per scoping dei dati e sicurezza, resta principalmente una preoccupazione applicativa/infrastrutturale.

In molti sistemi B2B, il tenant coincide con un’entità di dominio “Organizzazione”. In quel caso, la multi-tenancy è solo una forma di scoping per organizzazione, non una variabile magica.

4. DDD essenziale: Entità, Value Object, Aggregati

In DDD, l’aggregato è l’unità di coerenza transazionale. In multi-tenancy, un’altra regola pratica utile è: un aggregato non dovrebbe mai attraversare tenant. In altre parole, tutte le entità che appartengono a un aggregato devono essere “scopate” allo stesso tenant.

Modello di esempio: un sistema di ordini B2B in cui ogni organizzazione (tenant) gestisce clienti e ordini.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from typing import NewType
from uuid import UUID, uuid4

TenantId = NewType("TenantId", UUID)
OrderId = NewType("OrderId", UUID)
CustomerId = NewType("CustomerId", UUID)


@dataclass(frozen=True)
class Money:
    amount: Decimal
    currency: str = "EUR"

    def __post_init__(self) -> None:
        if self.amount < 0:
            raise ValueError("Money.amount non può essere negativo")


@dataclass
class OrderLine:
    sku: str
    quantity: int
    unit_price: Money

    def subtotal(self) -> Money:
        return Money(self.unit_price.amount * Decimal(self.quantity), self.unit_price.currency)


class Order:
    def __init__(self, tenant_id: TenantId, customer_id: CustomerId) -> None:
        self.id: OrderId = OrderId(uuid4())
        self.tenant_id = tenant_id
        self.customer_id = customer_id
        self.lines: list[OrderLine] = []
        self.created_at = datetime.utcnow()
        self._status: str = "DRAFT"

    @property
    def status(self) -> str:
        return self._status

    def add_line(self, line: OrderLine) -> None:
        if self._status != "DRAFT":
            raise ValueError("Non puoi modificare un ordine non DRAFT")
        if line.quantity <= 0:
            raise ValueError("La quantità deve essere > 0")
        self.lines.append(line)

    def total(self) -> Money:
        if not self.lines:
            return Money(Decimal("0.00"))
        currency = self.lines[0].unit_price.currency
        total = sum((l.subtotal().amount for l in self.lines), Decimal("0.00"))
        return Money(total, currency)

    def submit(self) -> None:
        if not self.lines:
            raise ValueError("Un ordine non può essere inviato senza righe")
        self._status = "SUBMITTED"

Notare: l’aggregato Order ha un tenant_id. Qui è lecito perché fa parte dell’identità di scoping: un ordine appartiene a un tenant. Evita però di propagare il tenant in ogni metodo se non serve. Molto spesso basta che l’aggregato lo contenga e che i repository lo usino per filtrare.

5. Tenant Context: portare il tenant dove serve (senza sporcare il dominio)

Il tenant viene tipicamente determinato a livello di richiesta (host, JWT claims, API key, path). Una volta ottenuto, va reso disponibile ai casi d’uso e ai repository. In Python, un approccio comune è usare contextvars per un contesto per-request, compatibile con async.

from __future__ import annotations

from contextvars import ContextVar
from dataclasses import dataclass
from typing import Optional
from uuid import UUID

_current_tenant: ContextVar[Optional[UUID]] = ContextVar("_current_tenant", default=None)


@dataclass(frozen=True)
class TenantContext:
    tenant_id: UUID


def set_current_tenant(tenant_id: UUID) -> None:
    _current_tenant.set(tenant_id)


def get_current_tenant() -> UUID:
    tenant_id = _current_tenant.get()
    if tenant_id is None:
        raise RuntimeError("Tenant non impostato nel contesto corrente")
    return tenant_id

Il dominio non deve conoscere contextvars. Il dominio riceve tenant_id quando crea aggregati, mentre l’infrastruttura (repository/DB) lo usa per filtrare.

6. Casi d’uso applicativi (Application Services)

Un application service coordina: input, autorizzazione, transazione, chiamate a repository e pubblicazione eventi. È il posto giusto per “legare” tenant context e dominio.

from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal
from uuid import UUID

# dipendenze astratte (porte)
class Orders:
    def add(self, order: "Order") -> None: ...
    def get(self, order_id: UUID) -> "Order | None": ...

class UnitOfWork:
    orders: Orders
    def commit(self) -> None: ...
    def rollback(self) -> None: ...
    def __enter__(self) -> "UnitOfWork": ...
    def __exit__(self, exc_type, exc, tb) -> None: ...


@dataclass(frozen=True)
class AddOrderLineCommand:
    order_id: UUID
    sku: str
    quantity: int
    unit_price: Decimal
    currency: str = "EUR"


class OrderService:
    def __init__(self, uow: UnitOfWork) -> None:
        self._uow = uow

    def create_order(self, customer_id: UUID) -> UUID:
        tenant_id = get_current_tenant()
        order = Order(tenant_id=tenant_id, customer_id=customer_id)
        with self._uow:
            self._uow.orders.add(order)
            self._uow.commit()
        return order.id

    def add_order_line(self, cmd: AddOrderLineCommand) -> None:
        with self._uow:
            order = self._uow.orders.get(cmd.order_id)
            if order is None:
                raise KeyError("Ordine non trovato")
            # difesa: l'ordine deve appartenere al tenant corrente
            if order.tenant_id != get_current_tenant():
                raise PermissionError("Accesso negato")
            order.add_line(
                OrderLine(
                    sku=cmd.sku,
                    quantity=cmd.quantity,
                    unit_price=Money(amount=cmd.unit_price, currency=cmd.currency),
                )
            )
            self._uow.commit()

Nota la “difesa in profondità”: anche se il repository filtra per tenant, l’application service può verificare lo scoping su operazioni sensibili. Non è sempre necessario, ma è utile in ambienti ad alto rischio.

7. Repository e Unit of Work: enforce del filtro tenant

Il repository è un adattatore tra dominio e persistenza. In multi-tenancy su tabelle condivise, il repository deve essere l’ultimo guardiano: ogni query deve includere il tenant, e ogni inserimento deve impostarlo.

Di seguito un esempio semplificato con SQLAlchemy (stile 2.0), usando una session per transazione.

from __future__ import annotations

from dataclasses import asdict
from typing import Optional
from uuid import UUID

from sqlalchemy import select
from sqlalchemy.orm import Session

# mapping ORM minimale (esempio)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, DateTime, ForeignKey
from datetime import datetime

class Base(DeclarativeBase):
    pass

class OrderRow(Base):
    __tablename__ = "orders"
    id: Mapped[UUID] = mapped_column(primary_key=True)
    tenant_id: Mapped[UUID] = mapped_column(index=True, nullable=False)
    customer_id: Mapped[UUID] = mapped_column(index=True, nullable=False)
    status: Mapped[str] = mapped_column(String(32), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)

class OrderLineRow(Base):
    __tablename__ = "order_lines"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    order_id: Mapped[UUID] = mapped_column(ForeignKey("orders.id"), index=True)
    tenant_id: Mapped[UUID] = mapped_column(index=True, nullable=False)
    sku: Mapped[str] = mapped_column(String(64), nullable=False)
    quantity: Mapped[int] = mapped_column(Integer, nullable=False)
    unit_price_amount: Mapped[str] = mapped_column(String(32), nullable=False)
    unit_price_currency: Mapped[str] = mapped_column(String(8), nullable=False)


class SqlAlchemyOrdersRepository:
    def __init__(self, session: Session) -> None:
        self._session = session

    def add(self, order: Order) -> None:
        tenant_id = get_current_tenant()
        if order.tenant_id != tenant_id:
            raise ValueError("Tenant mismatch: tentativo di salvare un aggregato di un altro tenant")

        self._session.add(
            OrderRow(
                id=order.id,
                tenant_id=order.tenant_id,
                customer_id=order.customer_id,
                status=order.status,
                created_at=order.created_at,
            )
        )
        for line in order.lines:
            self._session.add(
                OrderLineRow(
                    order_id=order.id,
                    tenant_id=order.tenant_id,
                    sku=line.sku,
                    quantity=line.quantity,
                    unit_price_amount=str(line.unit_price.amount),
                    unit_price_currency=line.unit_price.currency,
                )
            )

    def get(self, order_id: UUID) -> Optional[Order]:
        tenant_id = get_current_tenant()
        order_row = self._session.execute(
            select(OrderRow).where(OrderRow.id == order_id, OrderRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if order_row is None:
            return None

        # ricostruzione dell'aggregato (mapping inverso semplificato)
        order = Order(tenant_id=TenantId(order_row.tenant_id), customer_id=CustomerId(order_row.customer_id))
        order.id = OrderId(order_row.id)  # attenzione: qui aggiriamo l'init, in un progetto reale preferisci factory
        # carica righe
        lines = self._session.execute(
            select(OrderLineRow).where(OrderLineRow.order_id == order_id, OrderLineRow.tenant_id == tenant_id)
        ).scalars().all()
        for lr in lines:
            order.lines.append(
                OrderLine(
                    sku=lr.sku,
                    quantity=lr.quantity,
                    unit_price=Money(amount=Decimal(lr.unit_price_amount), currency=lr.unit_price_currency),
                )
            )
        return order

Due punti importanti:

  • Filtro tenant obbligatorio su ogni query. Non delegare al chiamante.
  • Tenant mismatch check in scrittura: se un bug prova a salvare un aggregato di un tenant diverso, meglio fallire subito.

8. Pattern di sicurezza: difesa in profondità

Il rischio principale di una multi-tenancy con tabelle condivise è un accesso non autorizzato a dati di un altro tenant. Riduci il rischio con più livelli:

  • Livello applicativo: tenant dal contesto, controlli autorizzativi (RBAC/ABAC), verifiche su comandi sensibili.
  • Livello repository: filtro tenant su query e scritture.
  • Livello database: vincoli, indici, e se possibile Row-Level Security (RLS).

8.1 Vincoli e indici

Con tabelle condivise, conviene progettare chiavi e vincoli pensando al tenant. Alcuni esempi:

  • Chiavi composte (tenant_id, id) oppure un id globale ma con vincoli di unicità che includono tenant.
  • Indici su (tenant_id, colonne di ricerca frequente) per evitare scansioni cross-tenant.

8.2 PostgreSQL Row-Level Security (RLS)

Se usi PostgreSQL, RLS permette di far rispettare al DB lo scoping tenant. Il principio è: l’app imposta una variabile di sessione (es. app.current_tenant) e le policy filtrano automaticamente.

-- Abilita RLS
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;

-- Policy: consenti accesso solo alle righe del tenant corrente
CREATE POLICY orders_tenant_isolation
ON orders
USING (tenant_id = current_setting('app.current_tenant')::uuid);

-- Ricorda di impostare la variabile per ogni connessione/sessione
-- SET app.current_tenant = '...';

In Python, devi impostare SET app.current_tenant all’inizio della transazione o quando acquisisci una connessione dal pool. Con SQLAlchemy puoi farlo in un hook (es. event.listen) o nel costruttore della Unit of Work.

from sqlalchemy import text

def set_tenant_on_connection(session: Session) -> None:
    tenant_id = get_current_tenant()
    session.execute(text("SET app.current_tenant = :t"), {"t": str(tenant_id)})

RLS non elimina la necessità del filtro a livello repository, ma riduce drasticamente l’impatto di un bug: anche se dimentichi un WHERE tenant_id = ..., il DB non restituirà righe di altri tenant.

9. Multi-tenancy e Bounded Context

In DDD, i bounded context delimitano modelli e linguaggi. La multi-tenancy può attraversarli, ma non necessariamente in modo identico. Esempi:

  • Nel contesto Billing, il tenant può avere piani, fatture, limiti.
  • Nel contesto Orders, il tenant “scopa” ordini e cataloghi.
  • Nel contesto Identity & Access, il tenant è centrale: utenti, ruoli, policy.

Quando un tenant cambia piano o policy, non vuoi che il dominio “Orders” dipenda direttamente da tabelle di “Billing”. Usa integrazioni tramite eventi o un Anti-Corruption Layer: “Orders” riceve un fatto già tradotto (es. TenantPlanChanged) e applica regole locali.

10. Eventi di dominio e integrazione per tenant

In un sistema multi-tenant, gli eventi devono includere lo scoping. È una buona pratica che ogni evento contenga il tenant, così puoi:

  • consumare eventi in modo isolato (code per tenant, partizioni, consumer group),
  • applicare policy di retention e audit per tenant,
  • eseguire replay mirati.
from dataclasses import dataclass
from uuid import UUID

@dataclass(frozen=True)
class DomainEvent:
    tenant_id: UUID

@dataclass(frozen=True)
class OrderSubmitted(DomainEvent):
    order_id: UUID
    total_amount: str
    currency: str

Evita di pubblicare eventi “globali” senza tenant, a meno che non rappresentino davvero un fatto globale (es. “nuova versione del software disponibile”).

11. Onboarding, migrazioni, e operatività

La multi-tenancy ha un costo operativo. DDD ti aiuta a mantenere il codice ordinato, ma devi anche progettare processi.

11.1 Onboarding tenant

  • Database per tenant: provisioning DB, credenziali, migrazioni iniziali.
  • Schema per tenant: creazione schema e migrazioni su schema.
  • Tabelle condivise: inserimento record “tenant” e configurazioni iniziali.

11.2 Migrazioni

Con schema-per-tenant o DB-per-tenant, le migrazioni devono iterare su più target. Serve automation, e spesso una strategia rolling. Con tabelle condivise, le migrazioni sono più semplici ma richiedono attenzione a indici e lock su tabelle grandi.

11.3 Backup e restore

Se prevedi restore per tenant, database-per-tenant è il più semplice. Con tabelle condivise, potresti dover estrarre solo le righe di un tenant o costruire strumenti ad hoc.

12. Testing: prevenire regressioni cross-tenant

Oltre ai test di dominio “puri”, aggiungi test che verificano l’isolamento. Un approccio efficace è costruire una suite che:

  • crea dati identici per due tenant diversi,
  • verifica che ogni query e caso d’uso non “veda” l’altro tenant.
import pytest
from uuid import uuid4

def test_order_isolation_by_tenant(uow_factory):
    tenant_a = uuid4()
    tenant_b = uuid4()

    # crea un ordine nel tenant A
    set_current_tenant(tenant_a)
    with uow_factory() as uow:
        service = OrderService(uow)
        order_id = service.create_order(customer_id=uuid4())

    # provi a leggerlo dal tenant B: deve fallire (None o PermissionError)
    set_current_tenant(tenant_b)
    with uow_factory() as uow:
        repo = uow.orders
        assert repo.get(order_id) is None

Se usi RLS, includi test di integrazione che verificano che il DB blocchi accessi non scopati, anche quando la query non filtra esplicitamente.

13. Osservabilità: tracciare per tenant

In produzione, quando qualcosa va storto, quasi sempre vuoi sapere “per quale tenant”. Integra il tenant in:

  • log (campo strutturato tenant_id),
  • tracing (tag/span attribute),
  • metriche (attenzione alle cardinalità: meglio raggruppare per tier/segmento se i tenant sono moltissimi),
  • audit (chi ha fatto cosa, quando, su quali risorse, per quale tenant).

Un antipattern: aggiungere il tenant come stringa in ogni messaggio di log non strutturato. Preferisci log JSON/strutturati, così puoi filtrare e correlare con facilità.

14. Checklist pratica

  • Il tenant è determinato in modo affidabile (JWT/host/API key) e validato.
  • Esiste un Tenant Context per-request, testabile e compatibile con async.
  • I repository applicano sempre il filtro tenant, e le scritture verificano mismatch.
  • Se possibile, usa protezioni a livello DB (vincoli, RLS, ruoli/permessi).
  • Gli eventi includono sempre il tenant (salvo eventi globali reali).
  • Test automatici coprono scenari cross-tenant e regressioni.
  • Logging/tracing includono tenant_id senza esplodere la cardinalità delle metriche.
  • Processi operativi (migrazioni, backup, onboarding) sono progettati insieme al modello.

Conclusione

Domain-Driven Design e multi-tenancy possono convivere bene se tieni separati i livelli: il dominio rimane focalizzato su invarianti e comportamenti, mentre la multi-tenancy viene applicata tramite contesto applicativo, repository rigorosi e difesa in profondità nel database.

Il risultato è un sistema più sicuro (meno fughe di dati), più manutenibile (meno inquinamento del modello) e più evolvibile (bounded context indipendenti, integrazioni tramite eventi).

Torna su