API Reference

This page is autogenerated from the source code's docstrings. Every public function, class, and constant in tenantshield.* is documented here.

tenantshield.context

tenantshield.context

Synchronous tenant context management for TenantShield.

This module defines :class:TenantContext and the free functions to activate, inspect, and construct contexts. The active context is stored in a module-level :class:contextvars.ContextVar, which makes it inherently thread- and asyncio-task-isolated without monkey-patching.

The asynchronous counterpart (atenant_scope) is added to this module in sub-phase 1A.4.

TenantId module-attribute

TenantId = NewType('TenantId', str)

Identifier for a tenant.

A NewType over str. At runtime it is exactly str; the wrapper exists only to give type checkers a way to distinguish tenant identifiers from arbitrary strings. Construct with TenantId("acme") (or TenantId(str(value)) when coercing from non-string sources).

Equality is by string value. No normalization (case folding, trimming) is applied; callers are responsible for normalizing at the system boundary.

TenantContext dataclass

A frozen, slotted tenant context.

Attributes:

Name	Type	Description
tenant_id	TenantId	The tenant identifier.
metadata	Mapping[str, object]	Arbitrary metadata associated with this context. Defaults to an empty mapping. Although the field is typed as Mapping (read-only in spirit), the underlying object may be a mutable dict — TenantShield does not enforce immutability of the contents. Treat the metadata as immutable from outside the package.

tenant_scope

tenant_scope(
    ctx: TenantContext,
) -> Generator[TenantContext, None, None]

Activate ctx as the current tenant for the duration of the block.

Parameters:

Name	Type	Description	Default
ctx	TenantContext	The tenant context to activate.	required

Yields:

Type	Description
TenantContext	The same context, for ergonomic with binding.

Example

with tenant_scope(bind_tenant(TenantId("acme"))) as ctx: ... assert current_tenant() is ctx

current_tenant

current_tenant() -> TenantContext

Return the active tenant context.

Raises:

Type	Description
MissingTenantContextError	if no tenant scope is active.

try_current_tenant

try_current_tenant() -> TenantContext | None

Return the active tenant context, or None if none is active.

bind_tenant

bind_tenant(
    tenant_id: TenantId, /, **metadata: object
) -> TenantContext

Construct a :class:TenantContext without activating it.

The / makes tenant_id positional-only — callers must write bind_tenant(TenantId("acme"), region="eu") and not bind_tenant(tenant_id=TenantId("acme"), region="eu"). This forces visual clarity at call sites.

Parameters:

Name	Type	Description	Default
tenant_id	TenantId	The tenant identifier (positional-only).	required
**metadata	object	Arbitrary metadata keyword arguments attached to the context.	{}

Returns:

Type	Description
TenantContext	A new TenantContext. To activate it, wrap with :func:tenant_scope.

Example

ctx = bind_tenant(TenantId("acme"), region="eu") with tenant_scope(ctx): ... pass

atenant_scope async

atenant_scope(
    ctx: TenantContext,
) -> AsyncGenerator[TenantContext, None]

Activate ctx as the current tenant for the duration of the async block.

Parameters:

Name	Type	Description	Default
ctx	TenantContext	The tenant context to activate.	required

Yields:

Type	Description
AsyncGenerator[TenantContext, None]	The same context, for ergonomic async with binding.

Example

async with atenant_scope(bind_tenant(TenantId("acme"))) as ctx: ... assert current_tenant() is ctx

tenantshield.exceptions

tenantshield.exceptions

Exception hierarchy for TenantShield.

All errors raised by TenantShield are subclasses of :class:TenantShieldError. The hierarchy is::

TenantShieldError
├── ConfigurationError
├── TenantContextError
│   ├── MissingTenantContextError
│   └── AmbiguousTenantContextError
├── EnforcementError
│   ├── CrossTenantAccessError
│   ├── UnscopedQueryError
│   └── CrossTenantJoinError
└── AdapterError

Errors with structured fields carry contextual information (tenant ids, model names, operation labels). They expose a :meth:to_dict method for serialization to audit sinks.

TenantShieldError

Bases: Exception

Base class for all TenantShield errors.

ConfigurationError

Bases: TenantShieldError

Raised when TenantShield is misconfigured or used in an unsupported way.

TenantContextError

Bases: TenantShieldError

Base class for tenant context problems.

EnforcementError

Bases: TenantShieldError

Base class for tenant isolation policy violations.

AdapterError

Bases: TenantShieldError

Base class for adapter-specific failures.

MissingTenantContextError dataclass

Bases: TenantContextError

Raised when an operation requires a tenant scope and none is active.

The auto-generated message includes a canonical-pattern hint pointing callers at the documented entry-point shapes (raw API + Django shortcut) per Finding #2 (Counterbook ADR-0015 catalog).

Parameters:

Name	Type	Description	Default
operation	str	A short label identifying the API or operation that needed the tenant context (e.g. "current_tenant", "query.all").	required
stack_context	Mapping[str, object]	Optional dictionary with additional debugging context.	_empty_metadata()

AmbiguousTenantContextError dataclass

Bases: TenantContextError

Raised when nested tenant scopes carry conflicting tenant identifiers.

Parameters:

Name	Type	Description	Default
tenant_id_outer	TenantId	Tenant active in the surrounding scope.	required
tenant_id_inner	TenantId	Tenant the inner scope attempted to activate.	required
stack_context	Mapping[str, object]	Optional dictionary with additional debugging context.	_empty_metadata()

CrossTenantAccessError dataclass

Bases: EnforcementError

Raised when an access crosses tenant boundaries.

Parameters:

Name	Type	Description	Default
tenant_id_expected	TenantId | None	The tenant that should have scoped the access.	required
tenant_id_actual	TenantId | None	The tenant actually carried by the data.	required
model	str | None	Name of the model or table involved.	required
operation	str	Short label identifying the operation (e.g. "read").	required
stack_context	Mapping[str, object]	Optional dictionary with additional debugging context.	_empty_metadata()

UnscopedQueryError dataclass

Bases: EnforcementError

Raised when a query touches a tenant-aware model without an active scope.

Parameters:

Name	Type	Description	Default
model	str | None	Name of the tenant-aware model that was queried.	required
operation	str	Short label identifying the operation (e.g. "filter").	required
stack_context	Mapping[str, object]	Optional dictionary with additional debugging context.	_empty_metadata()

CrossTenantJoinError dataclass

Bases: EnforcementError

Raised when a query joins models that belong to different tenants.

Parameters:

Name	Type	Description	Default
tenant_id_expected	TenantId | None	The tenant under which the join was attempted.	required
model_left	str	Name of the left side of the join.	required
model_right	str	Name of the right side of the join.	required
stack_context	Mapping[str, object]	Optional dictionary with additional debugging context.	_empty_metadata()

tenantshield.policies

tenantshield.policies

Tenant enforcement policies for TenantShield.

This module defines the :class:Policy protocol, the three :class:Decision variants (Allow, Deny, RequireScope), and three built-in policies (DenyByDefaultPolicy, AllowListPolicy, ChainPolicy). The :func:evaluate_and_audit helper composes a policy evaluation with an audit emission.

Policies are pure — they do not perform I/O or emit events by themselves. The audit emission is the responsibility of :func:evaluate_and_audit or of adapters in Phase 2+.

OperationType

Bases: StrEnum

Kind of operation being evaluated by a policy.

Inherits from str so it serializes cleanly to JSON.

Operation dataclass

Describes an operation to be evaluated by a policy.

Attributes:

Name	Type	Description
model	str	Fully-qualified model name (e.g. "app.Invoice").
operation_type	OperationType	Kind of operation.
tenant_context	TenantContext | None	The tenant context active during evaluation, or None if no context is active.
extras	Mapping[str, object]	Arbitrary additional data for custom policies.

Allow dataclass

Decision: allow the operation.

Note: slots=True is intentionally omitted because frozen=True + slots=True on a dataclass with no fields triggers a CPython bug (TypeError instead of FrozenInstanceError on setattr). Slots add no benefit to an empty class, so we keep this class non-slotted while Deny and RequireScope (which have fields) remain slotted.

Deny dataclass

Decision: deny the operation.

Attributes:

Name	Type	Description
reason	str	Human-readable reason. Surfaces in audit logs.

RequireScope dataclass

Decision: allow only if scoped by the given filter spec.

Attributes:

Name	Type	Description
filter_spec	Mapping[str, object]	Mapping of constraints. In Sub-phase 1B this is a free-form dict; adapters in Phase 2+ may impose structure.

Policy

Bases: Protocol

Contract for tenant enforcement policies.

Implementations should be stateless or stateful but thread-safe. Policies must not perform I/O — they receive the operation and return a decision synchronously.

DenyByDefaultPolicy dataclass

A policy that denies any operation without an active tenant context.

If a tenant context IS active, the operation is allowed. This is the minimal enforcement: prevent operations from running in the absence of explicit tenancy.

AllowListPolicy dataclass

A policy that allows operations only on a fixed set of models.

Operations on models outside the allowlist are denied regardless of tenant context. Operations on allowlisted models still require tenant context (delegate that to DenyByDefaultPolicy via composition).

Attributes:

Name	Type	Description
allowed_models	frozenset[str]	Frozen set of fully-qualified model names.

ChainPolicy dataclass

Compose multiple policies; first non-Allow decision wins.

Iterates the policies in order. If any returns Deny or RequireScope, the chain returns that decision (subsequent policies are not evaluated). Only if all policies return Allow does the chain return Allow.

Attributes:

Name	Type	Description
policies	tuple[Policy, ...]	Tuple of policies to apply in order.

evaluate_and_audit

evaluate_and_audit(
    policy: Policy, operation: Operation
) -> Decision

Evaluate operation against policy and emit an audit event.

Emits POLICY_ALLOW or POLICY_DENY based on the decision. RequireScope decisions emit POLICY_ALLOW (the scope is informational, not a denial).

Parameters:

Name	Type	Description	Default
policy	Policy	The policy to evaluate.	required
operation	Operation	The operation to evaluate.	required

Returns:

Type	Description
Decision	The decision returned by the policy.

tenantshield.audit

tenantshield.audit

Audit event bus for TenantShield.

This module defines the categorical event types, the immutable :class:AuditEvent record, and the :class:AuditSink protocol. The registry, the emit() dispatcher, and the built-in sinks are added in subsequent sub-tasks of Sub-phase 1B.

Subscribers ("sinks") consume :class:AuditEvent instances. Sinks must be thread- and async-safe and should not raise. The bus catches sink exceptions and emits SINK_FAILURE events to the remaining sinks.

AuditEventType

Bases: StrEnum

Categorical types of audit events.

Inherits from str so events serialize cleanly to JSON without a custom encoder.

AuditEvent dataclass

A single event emitted to the audit bus.

Attributes:

Name	Type	Description
event_type	AuditEventType	Categorical type of the event.
tenant_context	TenantContext | None	The tenant context active when the event occurred, or None if no context was active (e.g. SINK_FAILURE outside any scope).
payload	Mapping[str, object]	Arbitrary structured data. Treated as immutable from outside the package; mutating it does not break TenantShield invariants but is bad style.
timestamp	datetime	UTC timestamp at construction time. Generated by datetime.now(timezone.utc) per instance via default_factory.

AuditSink

Bases: Protocol

Contract for audit event consumers.

Implementations must be thread- and async-safe. Implementations should not raise exceptions; if they do, the bus catches them and emits a SINK_FAILURE event to the remaining sinks. Repeated failures from the same sink do not cause infinite recursion: the failure event is dispatched to sinks OTHER than the one that failed.

NullSink

A sink that discards all events. Useful for tests and as default.

InMemorySink

A sink that accumulates events in memory.

Not thread-safe by design for performance — intended for tests where the test controls concurrency. If you need thread-safety in production code, build a different sink.

Attributes:

Name	Type	Description
events	list[AuditEvent]	List of emitted events, in order.

clear

clear() -> None

Remove all accumulated events.

StructLogSink

An audit sink that forwards events to a structlog BoundLogger.

The sink does NOT configure structlog globally. The caller is responsible for the structlog processor chain. If no logger is provided, the sink uses structlog.get_logger("tenantshield.audit") which inherits the caller's global structlog configuration.

Parameters:

Name	Type	Description	Default
logger	BoundLogger | None	An optional pre-configured structlog logger. If None, a default logger bound to "tenantshield.audit" is used.	None

Example

import structlog from tenantshield.audit import StructLogSink, register_sink logger = structlog.get_logger("my_app").bind(component="tenancy") register_sink(StructLogSink(logger=logger))

register_sink

register_sink(sink: AuditSink) -> None

Register a sink to receive audit events.

Idempotent: registering the same sink twice (verified by identity) has no effect on the second call.

unregister_sink

unregister_sink(sink: AuditSink) -> None

Unregister a sink. If not registered, this is a no-op.

emit

emit(event: AuditEvent) -> None

Dispatch an event to all registered sinks.

Sinks that raise exceptions do not interrupt other sinks. For each failing sink, a SINK_FAILURE event is emitted to all OTHER registered sinks (preventing infinite recursion). A second-level failure (the SINK_FAILURE itself failing) is silently suppressed.

tenantshield.registry

tenantshield.registry

Tenant-aware model registry.

This module defines :class:RegistryEntry (metadata about a tenant-aware model) and :class:ModelRegistry (a thread-safe container of entries). The :data:default_registry instance and the module-level convenience functions are added in sub-task 1C.2.

RegistryEntry dataclass

Metadata about a tenant-aware model.

Attributes:

Name	Type	Description
model	type	The model class.
tenant_field	str	Name of the attribute/column that carries the tenant id.

ModelRegistry

Registry of tenant-aware models.

A registry maps model classes to their tenant metadata. Two registries are independent; users who need isolation construct their own instance. The package exposes a default_registry instance for the common case, and module-level convenience functions delegate to it (added in 1C.2).

Thread-safety: register/unregister/iteration operations are protected by an internal RLock. Read-only queries (is_registered, get) are also locked for consistency; the performance impact is negligible at registry sizes typical of real applications (hundreds of models).

register

register(
    model: type, *, tenant_field: str = "tenant_id"
) -> None

Register model as tenant-aware.

Idempotent: registering the same model twice with the same tenant_field is a no-op. Registering with a different tenant_field raises :class:ConfigurationError.

Parameters:

Name	Type	Description	Default
model	type	The model class to register.	required
tenant_field	str	Name of the attribute/column carrying the tenant id.	'tenant_id'

Raises:

Type	Description
ConfigurationError	if model is already registered with a different tenant_field.

unregister

unregister(model: type) -> None

Unregister model. If not registered, this is a no-op.

is_registered

is_registered(model: type) -> bool

Return True if model is registered as tenant-aware.

get

get(model: type) -> RegistryEntry

Return the :class:RegistryEntry for model.

Parameters:

Name	Type	Description	Default
model	type	The model class to look up.	required

Returns:

Name	Type	Description
The	RegistryEntry	class:RegistryEntry associated with model.

Raises:

Type	Description
ConfigurationError	if model is not registered.

clear

clear() -> None

Remove all registered models. Primarily for tests.

__iter__

__iter__() -> Iterator[RegistryEntry]

Iterate over registered entries.

Iteration takes a snapshot under lock; concurrent modifications during iteration do not affect the snapshot.

register_model

register_model(
    model: type | None = None,
    *,
    tenant_field: str = "tenant_id",
) -> type | Callable[[type], type]

Register a model as tenant-aware in the :data:default_registry.

Can be used as a decorator (with or without arguments) or called directly with the model class.

Parameters:

Name	Type	Description	Default
model	type | None	The model class to register. If None, returns a decorator.	None
tenant_field	str	Name of the attribute/column carrying the tenant id.	'tenant_id'

Returns:

Type	Description
type | Callable[[type], type]	The model class (when called with a model), or a decorator function
type | Callable[[type], type]	(when called without a model, e.g. @register_model(tenant_field=...)).

Examples:

>>> @register_model
... class Invoice:
...     tenant_id: str
...
>>> @register_model(tenant_field="org_id")
... class Org:
...     org_id: str
...
>>> register_model(LegacyModel, tenant_field="account_id")

is_tenant_aware

is_tenant_aware(model: type) -> bool

Return True if model is registered as tenant-aware in the default registry.

get_tenant_field

get_tenant_field(model: type) -> str

Return the tenant field name for model from the default registry.

Raises:

Type	Description
ConfigurationError	if model is not registered.