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")

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

Initialise the caching decorator.

Parameters

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

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

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``.

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

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.

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

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`

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)

def __init__(registry: ResolverRegistry) -> None

Initialise with a pre-populated resolver registry.

Parameters

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

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

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.

`ConfigOverridesConfig`

Configuration for the per-tenant config override layer.

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

`DatabaseIsolationStrategy`

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"

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

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.

async def remove_isolation(tenant_id: str) -> None

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

Parameters

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

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

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.

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

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`

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.

def __init__() -> None

Initialise empty tenant and config stores.

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

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``.

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

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``.

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

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.

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

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).

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

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.

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

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.

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

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.

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

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.

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

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``.

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

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).

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

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`

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 lexigram-sql’s DB context via TenantSQLContextBridge.

`IsolationStrategyRegistry`

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

Usage

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

def __init__() -> None

Initialise an empty registry.

def register(strategy: TenantIsolationStrategyProtocol) -> None

Parameters

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

def get(name: str) -> TenantIsolationStrategyProtocol

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.

def names() -> list[str]

Return the names of all registered strategies.

Returns

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

def with_defaults(cls) -> IsolationStrategyRegistry

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

Schema and database strategies are NOT included because they require lexigram-sql and environment-specific provisioning logic.

Returns

Type	Description
IsolationStrategyRegistry	A IsolationStrategyRegistry with ``"row_level"`` registered.

`LifecycleConfig`

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.

`ResolutionConfig`

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.

`ResolverRegistry`

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():
    ...

def __init__() -> None

Initialise an empty resolver registry.

def register(resolver: TenantResolverProtocol) -> None

Parameters

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

def ordered() -> list[TenantResolverProtocol]

Return resolvers sorted by ascending priority.

Returns

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

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

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`

Row-level isolation strategy.

Provisioning is a no-op because isolation is enforced at query time by lexigram-sql’s multi_tenant=True flag and TenantScope query scope rather than by separate schema/database resources.

Attributes: name: "row_level"

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

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).

async def remove_isolation(tenant_id: str) -> None

No-op.

Parameters

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

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

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.

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

No-op deprovisioning.

Parameters

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

Returns

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

`SchemaIsolationStrategy`

Creates a PostgreSQL schema per tenant.

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

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

Attributes: name: "schema"

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

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).

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

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.

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

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.

async def remove_isolation(tenant_id: str) -> None

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

Parameters

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

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

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`

Top-level tenancy configuration.

Loaded from the tenancy: key in application.yaml, with environment variable overrides via LEX_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`

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"]),
    )
))

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

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.

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

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`

Bundle provider that orchestrates all tenancy sub-providers.

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

TenantResolutionProvider
TenantLifecycleProvider
TenantConfigProvider
TenantIntegrationProvider

Usage

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

provider = TenancyProvider(TenancyConfig(...))

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

Initialise the bundle provider.

Parameters

Parameter	Type	Description
`config`	TenancyConfig \| None	Optional TenancyConfig. Defaults to all framework defaults when ``None``.

async def register(container: ContainerRegistrarProtocol) -> None

Delegate registration to all sub-providers.

Parameters

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

async def boot(container: BootContainerProtocol) -> None

Delegate boot to all sub-providers.

Parameters

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

async def shutdown() -> None

Delegate shutdown to sub-providers in reverse order.

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

Aggregate health across all sub-providers.

Returns

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

`TenantConfigService`

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")

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

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.

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

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``.

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

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.

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

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`

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

This middleware never rejects a request. It resolves the tenant if possible, sets TENANT_ID in the shared Context, and stores the TenantInfo in scope["state"]["tenant"] for downstream use by TenantGuard.

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

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

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`

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: ...

def __init__(ctx: Context) -> None

Initialise the guard with the shared context.

Parameters

Parameter	Type	Description
`ctx`	Context	The shared Context.

async def can_activate(execution_context: Any) -> bool

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`

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()

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

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.

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

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.

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

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.

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

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.

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

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.

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

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.

`TenantProvisioner`

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).

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

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.

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

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.

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

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`

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
    ...

def __init__(
    provider: TenantProviderProtocol,
    cache_ttl: int = 300
) -> None

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.

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

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.

def invalidate(tenant_id: str) -> None

Remove a specific tenant from the cache.

Parameters

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

def invalidate_all() -> None

Clear the entire validator cache.