API Reference

Classes

`CachedTenantConfigProvider`

Decorator that wraps any TenantConfigProviderProtocol with in-process TTL caching.

The entire config dict for a tenant is cached as a unit. A write (set_config) invalidates the cached dict for that tenant so the next read fetches fresh data.

Usage

base_provider = InMemoryTenantProvider()
cached = CachedTenantConfigProvider(base_provider, ttl=60)
value = await cached.get_config("tenant-abc", "feature_x")

base_provider = InMemoryTenantProvider()
cached = CachedTenantConfigProvider(base_provider, ttl=60)
value = await cached.get_config("tenant-abc", "feature_x")

__init__

def __init__(
    inner: TenantConfigProviderProtocol,
    ttl: int = 60
) -> None

source

Initialise the caching decorator.

Parameters

Parameter	Type	Description
`inner`	TenantConfigProviderProtocol	The underlying config provider to wrap.
`ttl`	int	Cache TTL in seconds.

get_config

async def get_config(
    tenant_id: str,
    key: str
) -> Any | None

source

Get a single config value, using the tenant-level cache.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose config is queried.
`key`	str	Configuration key.

Returns

Type	Description
Any \| None	The value if set, or ``None``.

get_all_config

async def get_all_config(tenant_id: str) -> dict[str, Any]

source

Get all config for a tenant, from cache if valid.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose config is retrieved.

Returns

Type	Description
dict[str, Any]	Dictionary of all key-value pairs for the tenant.

set_config

async def set_config(
    tenant_id: str,
    key: str,
    value: Any
) -> None

source

Set a config value and invalidate the cache.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose config is updated.
`key`	str	Configuration key.
`value`	Any	New value (must be JSON-serialisable).

`CompositeResolver`

source

Iterates the ResolverRegistry in priority order and returns the first non-``None`` result.

This is the primary resolution entry point used by TenantContextMiddleware.

Usage

resolver = CompositeResolver(registry)
tenant_id = await resolver.resolve(context)

resolver = CompositeResolver(registry)
tenant_id = await resolver.resolve(context)

__init__

def __init__(registry: ResolverRegistry) -> None

source

Initialise with a pre-populated resolver registry.

Parameters

Parameter	Type	Description
`registry`	ResolverRegistry	ResolverRegistry containing all configured resolvers.

resolve

async def resolve(context: TenantResolutionContext) -> str | None

source

Try each resolver in priority order and return the first result.

Parameters

Parameter	Type	Description
`context`	TenantResolutionContext	Immutable snapshot of the current request.

Returns

Type	Description
str \| None	The resolved ``tenant_id``, or ``None`` if no resolver could determine the tenant.

resolve_with_source

async def resolve_with_source(context: TenantResolutionContext) -> tuple[str, str] | None

source

Resolve the tenant and report which resolver won.

Reuses the existing registry priority order. The resolver name lets trust-tiered enforcement distinguish a server-verified source (jwt_claim) from client-influenced ones.

Parameters

Parameter	Type	Description
`context`	TenantResolutionContext	Immutable snapshot of the current request.

Returns

Type	Description
tuple[str, str] \| None	``(resolver_name, tenant_id)`` of the winning resolver, or ``None`` when no resolver could determine the tenant.

`ConfigOverridesConfig`

source

Configuration for the per-tenant config override layer.

Attributes: cache_ttl: Seconds a tenant’s config dict is cached in memory.

`DatabaseIsolationStrategy`

source

Full database isolation per tenant (reference implementation).

This is a reference implementation that applications must subclass and extend with their infrastructure-specific provisioning logic (e.g. AWS RDS, GCP Cloud SQL). It is NOT in IsolationStrategyRegistry.with_defaults().

Attributes: name: "database"

apply_isolation

async def apply_isolation(
    tenant_id: str,
    context: dict[str, Any]
) -> None

source

Record the tenant’s database backend in the execution context.

Parameters

Parameter	Type	Description
`tenant_id`	str	The active tenant.
`context`	dict[str, Any]	Mutable execution context dict.

remove_isolation

async def remove_isolation(tenant_id: str) -> None

source

No-op (connection routing resets with the request scope).

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose isolation context to remove.

provision_isolation

async def provision_isolation(tenant_id: str) -> Result[None, TenantError]

source

Provision a database for the tenant.

Must be overridden by application subclasses with infrastructure-specific database creation logic.

Parameters

Parameter	Type	Description
`tenant_id`	str	The newly created tenant.

Raises

Exception	Description
NotImplementedError	Always — subclasses must override this.

deprovision_isolation

async def deprovision_isolation(tenant_id: str) -> Result[None, TenantError]

source

Deprovision the tenant’s database.

Must be overridden by application subclasses.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant being deactivated.

Raises

Exception	Description
NotImplementedError	Always — subclasses must override this.

`InMemoryTenantProvider`

source

Dict-backed tenant store implementing both TenantProviderProtocol and TenantConfigProviderProtocol.

Suitable for unit testing, integration testing, and single-process development environments. Not suitable for production multi-process deployments (state is not shared across processes).

A single instance acts as both the tenant store and the config store, so only one registration is needed in the DI container.

__init__

def __init__() -> None

source

Initialise empty tenant and config stores.

get_tenant

async def get_tenant(tenant_id: str) -> TenantInfo | None

source

Return the tenant record for tenant_id, or None.

Parameters

Parameter	Type	Description
`tenant_id`	str	Unique tenant identifier.

Returns

Type	Description
TenantInfo \| None	TenantInfo or ``None``.

get_tenant_by_slug

async def get_tenant_by_slug(slug: str) -> TenantInfo | None

source

Return the tenant with the given slug, or None.

Parameters

Parameter	Type	Description
`slug`	str	URL-safe tenant identifier.

Returns

Type	Description
TenantInfo \| None	TenantInfo or ``None``.

list_tenants

async def list_tenants(
    *,
    active_only: bool = True
) -> list[TenantInfo]

source

List tenants, optionally filtering to active ones.

Parameters

Parameter	Type	Description
`active_only`	bool	Return only active tenants when ``True``.

Returns

Type	Description
list[TenantInfo]	List of TenantInfo.

create_tenant

async def create_tenant(command: CreateTenantCommand) -> Result[TenantInfo, TenantError]

source

Create and persist a new tenant record.

Parameters

Parameter	Type	Description
`command`	CreateTenantCommand	Creation parameters.

Returns

Type	Description
Result[TenantInfo, TenantError]	``Ok(TenantInfo)`` always (in-memory store never fails).

update_tenant

async def update_tenant(
    tenant_id: str,
    command: UpdateTenantCommand
) -> Result[TenantInfo, TenantError]

source

Update mutable fields on an existing tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to update.
`command`	UpdateTenantCommand	Fields to apply (``None`` fields are skipped).

Returns

Type	Description
Result[TenantInfo, TenantError]	``Ok(TenantInfo)`` on success, ``Err(TenantNotFoundError)`` if not found.

deactivate_tenant

async def deactivate_tenant(tenant_id: str) -> Result[None, TenantError]

source

Mark a tenant as inactive.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to deactivate.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success, ``Err(TenantNotFoundError)`` if not found.

activate_tenant

async def activate_tenant(tenant_id: str) -> Result[None, TenantError]

source

Mark a tenant as active.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to activate.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success, ``Err(TenantNotFoundError)`` if not found.

suspend_tenant

async def suspend_tenant(
    tenant_id: str,
    reason: str | None = None
) -> Result[None, TenantError]

source

Mark a tenant as suspended.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to suspend.
`reason`	str \| None	Optional reason string (stored in metadata for reference).

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success, ``Err(TenantNotFoundError)`` if not found.

get_config

async def get_config(
    tenant_id: str,
    key: str
) -> Any | None

source

Retrieve a single config value for a tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose config is queried.
`key`	str	Configuration key.

Returns

Type	Description
Any \| None	The value if set, or ``None``.

get_all_config

async def get_all_config(tenant_id: str) -> dict[str, Any]

source

Retrieve all config for a tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose config is retrieved.

Returns

Type	Description
dict[str, Any]	A copy of the tenant's config dict (empty dict if none set).

set_config

async def set_config(
    tenant_id: str,
    key: str,
    value: Any
) -> None

source

Set a config value for a tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose config is updated.
`key`	str	Configuration key.
`value`	Any	New value.

`IntegrationConfig`

source

Configuration for cross-package integration features.

Attributes: cache_key_prefix: When True, tenant-prefix cache keys via TenantCacheKeyDecorator. sql_context_bridge: When True, sync the core TENANT_ID context key into oridecon-sql’s DB context via TenantSQLContextBridge.

`IsolationStrategyRegistry`

source

Registry of TenantIsolationStrategyProtocol implementations keyed by ``strategy.name``.

Supports per-tenant strategy assignment via set_tenant_strategy and get_tenant_strategy for tier migration scenarios.

Usage

registry = IsolationStrategyRegistry.with_defaults()
strategy = registry.get("row_level")
await strategy.provision_isolation("tenant-abc")

registry = IsolationStrategyRegistry.with_defaults()
strategy = registry.get("row_level")
await strategy.provision_isolation("tenant-abc")

__init__

def __init__() -> None

source

Initialise an empty registry.

def register(strategy: TenantIsolationStrategyProtocol) -> None

source

Parameters

Parameter	Type	Description
`strategy`	TenantIsolationStrategyProtocol	Strategy object with a ``name`` attribute.

get

def get(name: str) -> TenantIsolationStrategyProtocol

source

Retrieve a strategy by name.

Parameters

Parameter	Type	Description
`name`	str	The strategy name (e.g. ``"row_level"``).

Returns

Type	Description
TenantIsolationStrategyProtocol	The registered strategy.

Raises

Exception	Description
TenantError	If no strategy with the given name is registered.

names

def names() -> list[str]

source

Return the names of all registered strategies.

Returns

Type	Description
list[str]	List of strategy name strings.

set_tenant_strategy

def set_tenant_strategy(
    tenant_id: str,
    strategy_name: str
) -> None

source

Assign an isolation strategy to a specific tenant.

Used during tier migration to cut over a tenant to its new isolation level. The strategy must already be registered.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant identifier.
`strategy_name`	str	The strategy name to assign.

Raises

Exception	Description
TenantError	If the strategy is not registered.

get_tenant_strategy

def get_tenant_strategy(
    tenant_id: str,
    default: str | None = None
) -> str | None

source

Return the strategy name assigned to a tenant, or default.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant identifier.
`default`	str \| None	Fallback value when no assignment exists.

Returns

Type	Description
str \| None	The assigned strategy name, or default.

with_defaults

def with_defaults(cls) -> IsolationStrategyRegistry

source

Create a registry pre-populated with the row-level strategy.

`LifecycleConfig`

source

Configuration for tenant lifecycle and provisioning.

Attributes: isolation_strategy: Name of the isolation strategy to use. Defaults to "row_level". auto_provision_isolation: When True, the provisioner runs the isolation strategy automatically on tenant creation.

`MigrationResult`

source

Outcome of a completed migration.

`MigrationServiceConfig`

source

Configuration for TenantMigrationService.

`ResolutionConfig`

source

Configuration for the tenant resolution chain.

Attributes: resolvers: Ordered list of resolver names to activate. Available names: "jwt_claim", "header", "subdomain", "path". header_name: HTTP header name read by HeaderTenantResolver. subdomain_pattern: Optional base domain for subdomain extraction (e.g. "app.com" → acme from acme.app.com). None disables subdomain resolver even if listed. path_pattern: Path pattern for PathTenantResolver. Use {tenant_id} as the placeholder. jwt_claim_key: JWT claim key read by JWTClaimTenantResolver. validator_cache_ttl: Seconds a validated TenantInfo is cached by TenantValidator. trusted_resolvers: Resolver names exempt from the membership cross-check because their source is server-verified. Defaults to ["jwt_claim"]. strict_membership: Default-deny gate. When True (default), a tenant resolved by a non-trusted resolver is bound only after the membership cross-check passes. Setting False reproduces the pre-fix behavior for migration only and is unsafe.

`ResolverRegistry`

source

Ordered registry of TenantResolverProtocol instances.

Resolvers are stored and returned in ascending priority order (lower number = tried first = higher trust level).

Usage

registry = ResolverRegistry()
registry.register(HeaderTenantResolver("x-tenant-id"))
for resolver in registry.ordered():
    ...

registry = ResolverRegistry()
registry.register(HeaderTenantResolver("x-tenant-id"))
for resolver in registry.ordered():
    ...

__init__

def __init__() -> None

source

Initialise an empty resolver registry.

def register(resolver: TenantResolverProtocol) -> None

source

Parameters

Parameter	Type	Description
`resolver`	TenantResolverProtocol	A resolver implementing TenantResolverProtocol.

ordered

def ordered() -> list[TenantResolverProtocol]

source

Return resolvers sorted by ascending priority.

Returns

Type	Description
list[TenantResolverProtocol]	List of resolvers in priority order (lowest number first).

from_config

def from_config(
    cls,
    resolver_names: list[str],
    header_name: str = 'x-tenant-id',
    subdomain_pattern: str | None = None,
    path_pattern: str | None = '/tenants/{tenant_id}/',
    jwt_claim_key: str = 'tenant_id'
) -> ResolverRegistry

source

Build a registry from a list of resolver names.

Only resolvers whose names appear in resolver_names are instantiated. Unknown names are silently ignored.

Parameters

Parameter	Type	Description
`resolver_names`	list[str]	Ordered list of resolver names to activate.
`header_name`	str	Header name for HeaderTenantResolver.
`subdomain_pattern`	str \| None	Base domain for SubdomainTenantResolver.
`path_pattern`	str \| None	Path pattern for PathTenantResolver.
`jwt_claim_key`	str	Claim key for JWTClaimTenantResolver.

Returns

Type	Description
ResolverRegistry	A populated ResolverRegistry.

`RowLevelIsolationStrategy`

source

Row-level isolation strategy (default).

Provisioning is a no-op because isolation is enforced at query time by oridecon-sql rather than by separate schema/database resources:

multi_tenant=True repositories (the preferred path) auto-scope every query to the ambient tenant and fail closed with TenantScopingError when no tenant is active.
TenantScope (declared per-repository via __scopes__) is an opt-in fallback for repositories that don’t use multi_tenant.

Attributes: name: "row_level"

apply_isolation

async def apply_isolation(
    tenant_id: str,
    context: dict[str, Any]
) -> None

source

No-op — isolation is handled at the ORM/query level.

Parameters

Parameter	Type	Description
`tenant_id`	str	The active tenant.
`context`	dict[str, Any]	Execution context dict (not modified).

remove_isolation

async def remove_isolation(tenant_id: str) -> None

source

No-op.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose isolation context to remove.

provision_isolation

async def provision_isolation(tenant_id: str) -> Result[None, TenantError]

source

No-op provisioning — nothing to create for row-level isolation.

Parameters

Parameter	Type	Description
`tenant_id`	str	The newly created tenant.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` always.

deprovision_isolation

async def deprovision_isolation(tenant_id: str) -> Result[None, TenantError]

source

No-op deprovisioning.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant being deactivated.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` always.

`SchemaIsolationStrategy`

source

Creates a PostgreSQL schema per tenant.

This is a reference implementation. Applications must register it explicitly — it is NOT in IsolationStrategyRegistry.with_defaults().

Requires oridecon-tenancy[sql]. Applications that need it must also configure the deprovision policy appropriate for their infrastructure.

Attributes: name: "schema"

__init__

def __init__(
    db_provider: DatabaseProviderProtocol,
    deprovision_policy: str = 'rename'
) -> None

source

Initialise the strategy.

Parameters

Parameter	Type	Description
`db_provider`	DatabaseProviderProtocol	Database provider implementing DatabaseProviderProtocol.
`deprovision_policy`	str	``"rename"`` (archive schema, default) or ``"drop"`` (permanently destroy).

provision_isolation

async def provision_isolation(tenant_id: str) -> Result[None, TenantError]

source

Create a PostgreSQL schema for the tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The newly created tenant.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success, ``Err(TenantProvisioningError)`` if the ``tenant_id`` contains characters that are unsafe for schema names.

apply_isolation

async def apply_isolation(
    tenant_id: str,
    context: dict[str, Any]
) -> None

source

Set the PostgreSQL search_path in the execution context.

Parameters

Parameter	Type	Description
`tenant_id`	str	The active tenant.
`context`	dict[str, Any]	Mutable execution context dict.

remove_isolation

async def remove_isolation(tenant_id: str) -> None

source

No-op (search_path resets with the DB connection).

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose isolation context to remove.

deprovision_isolation

async def deprovision_isolation(tenant_id: str) -> Result[None, TenantError]

source

Drop or rename the tenant schema.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant being deactivated.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success.

`TenancyConfig`

source

Top-level tenancy configuration.

Loaded from the tenancy: key in application.yaml, with environment variable overrides via ORI_TENANCY__* prefix.

Composed of four focused sub-configs.

Attributes: resolution: Resolver chain configuration. lifecycle: Lifecycle and isolation strategy configuration. overrides: Per-tenant config override layer configuration. integration: Cross-package integration feature toggles.

`TenancyModule`

source

Multi-tenant resolution, lifecycle, isolation, and configuration.

Use configure to register a fully configured tenancy stack. Use stub for test environments (in-memory store, header resolver only, no isolation provisioning).

Usage

app.add_module(TenancyModule.configure(
    config=TenancyConfig(
        resolution=ResolutionConfig(resolvers=["header"]),
    )
))

app.add_module(TenancyModule.configure(
    config=TenancyConfig(
        resolution=ResolutionConfig(resolvers=["header"]),
    )
))

configure

def configure(
    cls,
    config: TenancyConfig | None = None
) -> DynamicModule

source

Create a configured tenancy module.

Parameters

Parameter	Type	Description
`config`	TenancyConfig \| None	TenancyConfig or ``None`` for framework defaults.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: object = None
) -> DynamicModule

source

Create a test-friendly tenancy module.

Uses an in-memory store, header resolver only, no isolation provisioning, and no cache key prefix or SQL bridge.

Returns

Type	Description
DynamicModule	A DynamicModule for testing.

`TenancyProvider`

source

Bundle provider that orchestrates all tenancy sub-providers.

Mirrors the oridecon-auth AuthBundleProvider pattern. Delegates registration, boot, and shutdown to four focused sub-providers in order:

TenantResolutionProvider
TenantLifecycleProvider
TenantConfigProvider
TenantIntegrationProvider

Usage

from oridecon.tenancy.di.provider import TenancyProvider
from oridecon.tenancy.config import TenancyConfig

provider = TenancyProvider(TenancyConfig(...))

from oridecon.tenancy.di.provider import TenancyProvider
from oridecon.tenancy.config import TenancyConfig

provider = TenancyProvider(TenancyConfig(...))

__init__

def __init__(config: TenancyConfig | None = None) -> None

source

Initialise the bundle provider.

Parameters

Parameter	Type	Description
`config`	TenancyConfig \| None	Optional TenancyConfig. When ``None``, the orchestrator injects the typed ``tenancy`` yaml section after construction (``config_key``) and before register; framework defaults apply if no section exists.

async def register(container: ContainerRegistrarProtocol) -> None

source

Delegate registration to all sub-providers.

Late config binding: when configure() ran with no explicit config, the orchestrator injects the typed tenancy yaml section after construction and before this call; sub-providers are composed now so the automatic path behaves identically to the explicit one. An explicit constructor config always wins over any later assignment to config.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	The DI container registrar.

boot

async def boot(container: BootContainerProtocol) -> None

source

Delegate boot to all sub-providers.

Parameters

Parameter	Type	Description
`container`	BootContainerProtocol	The DI container for boot phase.

shutdown

async def shutdown() -> None

source

Delegate shutdown to sub-providers in reverse order.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Aggregate health across all sub-providers.

Returns

Type	Description
HealthCheckResult	A HealthCheckResult reflecting the worst sub-provider status.

`TenantConfigService`

source

High-level per-tenant configuration service.

Combines a raw TenantConfigProviderProtocol backend with a defaults dict and event emission.

Priority order for get:

Tenant-specific override from the provider.
Application default from defaults.
None.

Usage

service = TenantConfigService(provider, defaults={"max_users": 50}, event_bus=bus)
value = await service.get("tenant-abc", "max_users")

service = TenantConfigService(provider, defaults={"max_users": 50}, event_bus=bus)
value = await service.get("tenant-abc", "max_users")

__init__

def __init__(
    config_provider: TenantConfigProviderProtocol,
    defaults: dict[str, Any],
    event_bus: DomainEventPublisherProtocol
) -> None

source

Initialise the service.

Parameters

Parameter	Type	Description
`config_provider`	TenantConfigProviderProtocol	Underlying key-value config backend.
`defaults`	dict[str, Any]	Application-level default values used when no tenant override is set.
`event_bus`	DomainEventPublisherProtocol	Event bus for publishing TenantConfigChanged events.

get

async def get(
    tenant_id: str,
    key: str
) -> Any

source

Get a config value with fallback to defaults.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose config is queried.
`key`	str	Configuration key.

Returns

Type	Description
Any	The tenant override if set, otherwise the application default, otherwise ``None``.

set

async def set(
    tenant_id: str,
    key: str,
    value: Any
) -> None

source

Set a config value and publish the change event.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose config is updated.
`key`	str	Configuration key.
`value`	Any	New value.

get_effective_config

async def get_effective_config(tenant_id: str) -> dict[str, Any]

source

Merge defaults with tenant overrides (tenant wins on conflict).

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose effective config is computed.

Returns

Type	Description
dict[str, Any]	A merged dict where tenant overrides take precedence over defaults.

`TenantContextMiddleware`

source

ASGI middleware that resolves the tenant for every HTTP/WebSocket request.

This middleware never rejects a request; it refuses to bind a tenant it cannot verify. It resolves the tenant, authorizes the binding against the caller’s identity (server-verified resolvers bind directly; client-influenced ones require a membership cross-check), then sets TENANT_ID in the shared Context and stores the TenantInfo in scope["state"]["tenant"] for downstream use by TenantGuard. The context token is always reset on exit.

Registration order: after RequestContextMiddleware and DIScopeMiddleware, before application middleware.

__init__

def __init__(
    app: ASGIApp,
    resolver: Any,
    validator: Any,
    ctx: Context
) -> None

source

Initialise the middleware.

Parameters

Parameter	Type	Description
`app`	ASGIApp	The next ASGI application in the chain.
`resolver`	Any	CompositeResolver instance.
`validator`	Any	TenantValidator instance.
`ctx`	Context	The shared Context.

`TenantGuard`

source

Route-level guard that enforces tenant context presence.

Applied via @use_guards(TenantGuard) on controllers or individual routes. Returns False (triggering a 403 response) if no active tenant is present in the current request scope.

The guard reads TENANT_ID from the shared context AND verifies the tenant stored in scope["state"]["tenant"] has ACTIVE status.

Usage

@use_guards(TenantGuard)
class TenantAwareController: ...

@use_guards(TenantGuard)
class TenantAwareController: ...

__init__

def __init__(ctx: Context) -> None

source

Initialise the guard with the shared context.

Parameters

Parameter	Type	Description
`ctx`	Context	The shared Context.

can_activate

async def can_activate(execution_context: Any) -> bool

source

Check whether the current request may proceed.

Parameters

Parameter	Type	Description
`execution_context`	Any	The framework execution context, which must expose ``execution_context.request.scope`` for tenant state lookup.

Returns

Type	Description
bool	``True`` if a valid, active tenant is in context; ``False`` otherwise (results in 403 Forbidden).

`TenantLifecycleService`

source

Orchestrates tenant CRUD operations with event emission and cache invalidation.

All mutations publish the corresponding domain event via the event bus and invalidate the TenantValidator cache so subsequent requests see fresh state.

Usage

service = TenantLifecycleService(provider, provisioner, event_bus, validator)
result = await service.create_tenant(CreateTenantCommand(slug="acme", name="ACME Corp"))
if result.is_ok():
    tenant = result.unwrap()

service = TenantLifecycleService(provider, provisioner, event_bus, validator)
result = await service.create_tenant(CreateTenantCommand(slug="acme", name="ACME Corp"))
if result.is_ok():
    tenant = result.unwrap()

__init__

def __init__(
    provider: TenantProviderProtocol,
    provisioner: TenantProvisioner,
    event_bus: DomainEventPublisherProtocol,
    validator: TenantValidator
) -> None

source

Initialise the service.

Parameters

Parameter	Type	Description
`provider`	TenantProviderProtocol	Tenant storage backend.
`provisioner`	TenantProvisioner	Isolation provisioner for new/removed tenants.
`event_bus`	DomainEventPublisherProtocol	Event bus for publishing domain events.
`validator`	TenantValidator	Validator cache to invalidate on mutations.

create_tenant

async def create_tenant(command: CreateTenantCommand) -> Result[TenantInfo, TenantError]

source

Create a new tenant, provision isolation, and publish the event.

Parameters

Parameter	Type	Description
`command`	CreateTenantCommand	Tenant creation parameters.

Returns

Type	Description
Result[TenantInfo, TenantError]	``Ok(TenantInfo)`` on success, ``Err(TenantError)`` on failure.

update_tenant

async def update_tenant(
    tenant_id: str,
    command: UpdateTenantCommand
) -> Result[TenantInfo, TenantError]

source

Update a tenant’s mutable fields.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to update.
`command`	UpdateTenantCommand	Fields to apply.

Returns

Type	Description
Result[TenantInfo, TenantError]	``Ok(TenantInfo)`` with the updated record, ``Err(TenantError)`` on failure.

deactivate_tenant

async def deactivate_tenant(tenant_id: str) -> Result[None, TenantError]

source

Deactivate a tenant and publish the event.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to deactivate.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success, ``Err(TenantError)`` on failure.

activate_tenant

async def activate_tenant(tenant_id: str) -> Result[None, TenantError]

source

Activate a tenant and publish the event.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to activate.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success, ``Err(TenantError)`` on failure.

suspend_tenant

async def suspend_tenant(
    tenant_id: str,
    reason: str | None = None
) -> Result[None, TenantError]

source

Suspend a tenant and publish the event.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to suspend.
`reason`	str \| None	Optional reason for the suspension.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success, ``Err(TenantError)`` on failure.

`TenantMigrationService`

source

Public API for tenant tier migrations.

Builds and executes TenantTierMigrationSaga instances, publishes lifecycle events, and returns structured results.

Usage

result = await migration_service.migrate_tier(
    tenant_id="tenant-abc",
    target_tier="m5",
)
if result.is_ok():
    print(result.unwrap())

result = await migration_service.migrate_tier(
    tenant_id="tenant-abc",
    target_tier="m5",
)
if result.is_ok():
    print(result.unwrap())

__init__

def __init__(
    tenant_provider: TenantProviderProtocol,
    isolation_registry: IsolationStrategyRegistry,
    config_service: TenantConfigService,
    write_pause_registry: WritePauseRegistry,
    checkpoint_store: ContentCheckpointStoreProtocol,
    copy_strategy: MigrationCopyStrategy,
    event_bus: DomainEventPublisherProtocol | None = None,
    config: MigrationServiceConfig | None = None
) -> None

source

Initialise the service.

Parameters

Parameter	Type	Description
`tenant_provider`	TenantProviderProtocol	Tenant storage backend.
`isolation_registry`	IsolationStrategyRegistry	Registry of isolation strategies.
`config_service`	TenantConfigService	Per-tenant config overrides.
`write_pause_registry`	WritePauseRegistry	Write-pause coordination.
`checkpoint_store`	ContentCheckpointStoreProtocol	Content-addressed checkpoint backend.
`copy_strategy`	MigrationCopyStrategy	Strategy for copying tenant data.
`event_bus`	DomainEventPublisherProtocol \| None	Optional event bus for lifecycle events.
`config`	MigrationServiceConfig \| None	Optional service configuration.

migrate_tier

async def migrate_tier(
    tenant_id: str,
    target_tier: str,
    *,
    actor_id: str | None = None
) -> Result[MigrationResult, TenantError]

source

Migrate a tenant to a new isolation tier.

This is the main entry point. It validates the request, builds a TenantTierMigrationSaga, executes it, and publishes lifecycle events.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant to migrate.
`target_tier`	str	Target tier name (e.g. ``"m5"``, ``"m6"``).
`actor_id`	str \| None	Optional identifier of the user / system that triggered the migration.

Returns

Type	Description
Result[MigrationResult, TenantError]	``Ok(MigrationResult)`` on success, ``Err(MigrationError)`` on failure.

`TenantProvisioner`

source

Orchestrates data isolation setup and teardown when tenants are created or deactivated.

When auto_provision is False (e.g. in tests), both provision and deprovision are no-ops that immediately return Ok(None).

__init__

def __init__(
    strategy: TenantIsolationStrategyProtocol,
    auto_provision: bool = True
) -> None

source

Initialise the provisioner.

Parameters

Parameter	Type	Description
`strategy`	TenantIsolationStrategyProtocol	The isolation strategy to use for provisioning.
`auto_provision`	bool	When ``False``, skip provisioning entirely. Useful for testing.

provision

async def provision(tenant_id: str) -> Result[None, TenantError]

source

Provision isolation resources for a new tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The newly created tenant.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success (or when ``auto_provision=False``), ``Err(TenantError)`` on failure.

deprovision

async def deprovision(tenant_id: str) -> Result[None, TenantError]

source

Deprovision isolation resources for a deactivated tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant being deactivated.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)`` on success (or when ``auto_provision=False``), ``Err(TenantError)`` on failure.

`TenantValidator`

source

Validates a ``tenant_id`` against the tenant store with TTL caching.

Caches the lookup result in an in-process dict. In multi-process deployments the cache is process-local; the TTL bounds staleness.

Usage

validator = TenantValidator(provider, cache_ttl=300)
info = await validator.validate("tenant-abc")
if info:
    # tenant is active
    ...

validator = TenantValidator(provider, cache_ttl=300)
info = await validator.validate("tenant-abc")
if info:
    # tenant is active
    ...

__init__

def __init__(
    provider: TenantProviderProtocol,
    cache_ttl: int = 300,
    membership: TenantMembershipProtocol | None = None,
    trusted_resolvers: list[str] | None = None,
    strict_membership: bool = True
) -> None

source

Initialise the validator.

Parameters

Parameter	Type	Description
`provider`	TenantProviderProtocol	Tenant storage implementing TenantProviderProtocol.
`cache_ttl`	int	Seconds to cache a validated TenantInfo record. Defaults to 300 seconds.
`membership`	TenantMembershipProtocol \| None	Optional membership check. When ``None`` and ``strict_membership`` is ``True``, non-trusted resolvers never bind a tenant (default-deny).
`trusted_resolvers`	list[str] \| None	Resolver names exempt from the membership check because their source is server-verified. Defaults to ``["jwt_claim"]``.
`strict_membership`	bool	When ``True`` (default), deny tenant binding when a non-trusted resolver's tenant cannot be verified against the caller's identity. Setting ``False`` reproduces the pre-fix behavior and is unsafe — it is a migration-only lever.

validate

async def validate(tenant_id: str) -> TenantInfo | None

source

Validate that a tenant exists and is active.

Results are cached for cache_ttl seconds. Returns None for inactive, suspended, or provisioning tenants.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to validate.

Returns

Type	Description
TenantInfo \| None	The TenantInfo if the tenant is active, or ``None`` otherwise.

invalidate

def invalidate(tenant_id: str) -> None

source

Remove a specific tenant from the cache.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose cache entry should be evicted.

invalidate_all

def invalidate_all() -> None

source

Clear the entire validator cache.

authorize

async def authorize(
    *,
    resolver_name: str,
    user_id: str | None,
    tenant_id: str
) -> bool

source

Decide whether tenant_id may be bound to the caller.

Trusted resolvers are server-verified and bind without further checks. Any other resolver must pass the membership cross-check against user_id; when strict and verification is impossible (no membership protocol bound, or no user_id), the tenant is refused — never bound unverified (default-deny).

Parameters

Parameter	Type	Description
`resolver_name`	str	Name of the resolver that won the chain.
`user_id`	str \| None	Authenticated identity from ``scope["state"]["user_id"]`` (``None`` for anonymous).
`tenant_id`	str	The tenant the resolver produced.

Returns

Type	Description
bool	``True`` if the tenant may be bound, ``False`` otherwise.

`WritePauseRegistry`

source

Coordinates write-pausing during tenant tier migration.

Uses an optional CacheBackendProtocol for distributed deployments; falls back to an in-memory dict.

Services check is_paused before accepting write operations for a tenant undergoing migration.

__init__

def __init__(backend: CacheBackendProtocol | None = None) -> None

source

Initialise the registry.

Parameters

Parameter	Type	Description
`backend`	CacheBackendProtocol \| None	Optional distributed cache backend.

pause

async def pause(
    tenant_id: str,
    reason: str = 'migration'
) -> None

source

Pause writes for a tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant to pause.
`reason`	str	Human-readable reason for the pause.

resume

async def resume(tenant_id: str) -> None

source

Resume writes for a tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant to resume.

is_paused

async def is_paused(tenant_id: str) -> bool

source

Check whether writes are paused for a tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant to check.

Returns

Type	Description
bool	``True`` if writes are paused.

pause_reason

async def pause_reason(tenant_id: str) -> str | None

source

Return the reason writes are paused, or None.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant to check.