API Reference

Protocols

`CacheBackendProtocol`

Protocol for cache backend implementations.

This interface defines the contract that all cache backend implementations must follow. It provides a unified API for different storage mechanisms (memory, Redis, Memcached, etc.).

Example

class RedisBackend:
    async def get(self, key: str) -> Result[Any | None, CacheError]:
        try:
            data = await self._redis.get(key)
            return Ok(self._serializer.deserialize(data) if data else None)
        except Exception as exc:
            return Err(CacheError(str(exc)))

    async def set(
        self, key: str, value: Any, ttl: int | None = None
    ) -> Result[None, CacheError]:
        try:
            data = self._serializer.serialize(value)
            await self._redis.set(key, data, ex=ttl)
            return Ok(None)
        except Exception as exc:
            return Err(CacheError(str(exc)))

get

async def get(key: str) -> Result[Any | None, CacheError]

source

Get a value from the cache.

Parameters

Parameter	Type	Description
`key`	str	The cache key to retrieve.

Returns

Type	Description
Result[Any \| None, CacheError]	Ok(value) if found, Ok(None) if not found, Err(CacheError) on failure.

Note

The return type is Any by design: the cache is a heterogeneous store and the caller knows the expected type. Use cast(T, result.unwrap()) at call sites for type safety.

set

async def set(
    key: str,
    value: Any,
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set a value in the cache with optional TTL.

Parameters

Parameter	Type	Description
`key`	str	The cache key to set.
`value`	Any	The value to cache.
`ttl`	int \| None	Time to live in seconds (optional).

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

delete

async def delete(key: str) -> Result[bool, CacheError]

source

Delete a value from the cache.

Parameters

Parameter	Type	Description
`key`	str	The cache key to delete.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if deleted, Ok(False) if not found, Err(CacheError) on failure.

delete_many

async def delete_many(keys: list[str]) -> Result[int, CacheError]

source

Delete multiple values from the cache.

Parameters

Parameter	Type	Description
`keys`	list[str]	List of cache keys to delete.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

delete_pattern

async def delete_pattern(pattern: str) -> Result[int, CacheError]

source

Delete all keys matching a glob-style pattern.

Parameters

Parameter	Type	Description
`pattern`	str	Glob pattern (e.g. ``"pet:list:"``). Supports ```` as a wildcard matching any sequence of characters.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

exists

async def exists(key: str) -> Result[bool, CacheError]

source

Check if a key exists in the cache.

Parameters

Parameter	Type	Description
`key`	str	The cache key to check.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if exists, Ok(False) otherwise, Err(CacheError) on failure.

clear

async def clear() -> Result[None, CacheError]

source

Clear all values from the cache.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

get_many

async def get_many(keys: list[str]) -> Result[dict[str, Any], CacheError]

source

Get multiple values from the cache.

Parameters

Parameter	Type	Description
`keys`	list[str]	List of cache keys to retrieve.

Returns

Type	Description
Result[dict[str, Any], CacheError]	Ok(dict) mapping found keys to values, Err(CacheError) on failure.

set_many

async def set_many(
    items: dict[str, Any],
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set multiple values in the cache.

Parameters

Parameter	Type	Description
`items`	dict[str, Any]	Dictionary of key-value pairs to cache.
`ttl`	int \| None	Time to live in seconds for all items.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if all items set successfully, Err(CacheError) on failure.

health_check

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

source

Perform a health check on the cache backend.

Returns

Type	Description
HealthCheckResult	Structured HealthCheckResult.

`ClockProtocol`

source

Injectable time source for the framework.

now

def now() -> datetime

source

Return the current timezone-aware UTC time.

monotonic

def monotonic() -> float

source

Return a monotonic elapsed-time counter.

timestamp

def timestamp() -> float

source

Return the current Unix timestamp in UTC seconds.

time

def time() -> float

source

Backward-compatible alias for timestamp.

`ConfigProtocol`

source

Protocol for configuration access across the framework.

Any configuration object (BaseConfig, LexigramConfig, or custom implementations) can satisfy this protocol by implementing get(), get_section(), and has_section().

Example

config = container.resolve(ConfigProtocol)
db_url = config.get("database.url", "sqlite:///default.db")
logging = config.get_section("logging", LoggingConfig)

config = container.resolve(ConfigProtocol)
db_url = config.get("database.url", "sqlite:///default.db")
logging = config.get_section("logging", LoggingConfig)

environment

property environment() -> Environment

source

The active deployment environment.

is_production

property is_production() -> bool

source

Return True when the active environment is production.

is_development

property is_development() -> bool

source

Return True when the active environment is development.

is_testing

property is_testing() -> bool

source

Return True when the active environment is testing.

is_staging

property is_staging() -> bool

source

Return True when the active environment is staging.

is_debug

property is_debug() -> bool

source

Return True when debug mode is enabled.

get

def get(
    key: str,
    default: Any = None
) -> Any

source

Get a configuration value by dot-notation key.

Parameters

Parameter	Type	Description
`key`	str	Configuration key (e.g. ``"app.name"`` or ``"database.url"``).
`default`	Any	Value returned when the key is not found.

Returns

Type	Description
Any	The configuration value, or default if not found.

get_section

def get_section(
    name: str,
    model_cls: type[T] | None = None
) -> T | dict[str, Any]

source

Get a typed configuration section.

Parameters

Parameter	Type	Description
`name`	str	Section name (e.g. ``"logging"``).
`model_cls`	type[T] \| None	Optional model class to coerce the section into.

Returns

Type	Description
T \| dict[str, Any]	A model instance when model_cls is provided, otherwise a raw dict or the attribute value.

has_section

def has_section(name: str) -> bool

source

Check whether a configuration section exists.

Parameters

Parameter	Type	Description
`name`	str	Section name to check.

Returns

Type	Description
bool	True if the section is present.

`EventMiddlewareProtocol`

source

Protocol for event bus middleware.

Middleware intercepts event publication, allowing cross-cutting concerns like logging, metrics, or error handling to be applied transparently.

The middleware receives the event and a next_handler coroutine that invokes the next middleware (or the actual handlers). The middleware must call next_handler to continue the chain.

Example

class LoggingMiddleware:
    async def __call__(self, event: Any, next_handler: Any) -> None:
        logger.info("publishing", event_type=type(event).__name__)
        await next_handler(event)
        logger.info("published", event_type=type(event).__name__)

class LoggingMiddleware:
    async def __call__(self, event: Any, next_handler: Any) -> None:
        logger.info("publishing", event_type=type(event).__name__)
        await next_handler(event)
        logger.info("published", event_type=type(event).__name__)

`HookRegistryProtocol`

source

Structural protocol for hook registries.

Defines the contract for action hooks (fire-and-forget, error-isolated) and filter hooks (value-pipeline, error-propagating) with priority ordering.

register_action

def register_action(
    hook_name: str,
    handler: Callable[Ellipsis, Any],
    priority: int = 100,
    *,
    once: bool = False
) -> None

source

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.
`handler`	Callable[Ellipsis, Any]	Callable to invoke. Can be sync or async.
`priority`	int	Lower values run first. Defaults to 100.
`once`	bool	If True, auto-remove handler after first invocation.

register_filter

def register_filter(
    hook_name: str,
    handler: Callable[Ellipsis, Any],
    priority: int = 100,
    *,
    once: bool = False
) -> None

source

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.
`handler`	Callable[Ellipsis, Any]	Callable that receives and returns a value.
`priority`	int	Lower values run first. Defaults to 100.
`once`	bool	If True, auto-remove handler after first invocation.

unregister_action

def unregister_action(
    hook_name: str,
    handler: Callable[Ellipsis, Any]
) -> bool

source

Remove an action handler. Returns True if found and removed.

unregister_filter

def unregister_filter(
    hook_name: str,
    handler: Callable[Ellipsis, Any]
) -> bool

source

Remove a filter handler. Returns True if found and removed.

call_action

async def call_action(
    hook_name: str,
    **kwargs: Any
) -> None

source

Invoke all action handlers. Errors are isolated and logged.

apply_filter

async def apply_filter(
    hook_name: str,
    value: Any,
    **kwargs: Any
) -> Any

source

Pass value through all filter handlers. Errors propagate.

has_action

def has_action(hook_name: str) -> bool

source

Check if any action handlers are registered.

has_filter

def has_filter(hook_name: str) -> bool

source

Check if any filter handlers are registered.

clear

def clear(hook_name: str | None = None) -> None

source

Clear handlers for a specific hook, or all hooks.

`IdGeneratorProtocol`

source

Injectable identity generator.

strategy

property strategy() -> IdStrategy

source

Return the active ID generation strategy.

generate

def generate() -> str

source

Generate a new unique identifier.

generate_for

def generate_for(entity_type: str) -> str

source

Generate a new identifier for a specific entity type.

Classes

`AbstractBuilder`

source

Base class for all fluent builders in Lexigram.

Provides foundational support for state management and fluent chaining.

__init__

def __init__() -> None

source

Initialize builder state.

copy

def copy() -> Self

source

Return a deep copy of the builder state.

Useful for creating slightly different variations of an object without rebuilding from scratch.

build

def build() -> T

source

Construct the final object from the builder state.

Returns

Type	Description
T	The constructed object of type T.

`AppState`

source

Application lifecycle states.

`Application`

source

The composition root.

Usage

async with Application.boot(providers=[MyProvider()]) as app:
    # resolve Invoker from the container for entry-point injection
    from lexigram.app.invoker import Invoker
    invoker = await app.container.resolve(Invoker)
    await invoker.invoke(main)

async with Application.boot(providers=[MyProvider()]) as app:
    # resolve Invoker from the container for entry-point injection
    from lexigram.app.invoker import Invoker
    invoker = await app.container.resolve(Invoker)
    await invoker.invoke(main)

Or for long-running processes

app = Application()
app.add_provider(MyProvider())
from lexigram.app import run_application
await run_application(app)

app = Application()
app.add_provider(MyProvider())
from lexigram.app import run_application
await run_application(app)

__init__

def __init__(
    name: str = 'lexigram-app',
    config: LexigramConfig | None = None
) -> None

source

state

property state() -> AppState

source

Current application state.

is_running

property is_running() -> bool

source

Check if application is running.

logger

property logger() -> LoggerProtocol

source

Get the application logger.

config

property config() -> LexigramConfig

source

Application configuration.

container

property container() -> Container

source

DI container.

providers

property providers() -> list[Provider]

source

Get all registered providers.

register_secrets

def register_secrets(
    secrets: dict[str, str],
    *,
    policy: Any | None = None
) -> None

source

The framework validates these secrets during start, before any provider’s register() is called. The app declares what to validate; the framework supplies the logic via SecretsValidator.

Parameters

Parameter	Type	Description
`secrets`	dict[str, str]	Mapping of secret name → value. Names should be descriptive (e.g. ``"JWT_SECRET"``), as they appear in error messages.
`policy`	Any \| None	Optional SecretsPolicy that controls strictness. When ``None`` a default policy is derived from the active Environment.

Example

app.register_secrets(
    {
        "JWT_SECRET": os.environ["JWT_SECRET"],
        "OAUTH_CLIENT_SECRET": os.environ["OAUTH_CLIENT_SECRET"],
    },
)

app.register_secrets(
    {
        "JWT_SECRET": os.environ["JWT_SECRET"],
        "OAUTH_CLIENT_SECRET": os.environ["OAUTH_CLIENT_SECRET"],
    },
)

register_secrets_from_store

def register_secrets_from_store(
    store: Any,
    secret_names: list[str],
    *,
    policy: Any | None = None
) -> None

source

The framework validates these secrets during start, before any provider’s register() is called. Secrets are retrieved from the store at validation time and validated using the same SecretsValidator logic as literal secrets.

Parameters

Parameter	Type	Description
`store`	Any	An implementation of SecretStoreProtocol.
`secret_names`	list[str]	List of secret names to retrieve from the store. Names should be descriptive (e.g. ``"JWT_SECRET"``), as they appear in error messages.
`policy`	Any \| None	Optional SecretsPolicy that controls strictness. When ``None`` a default policy is derived from the active Environment.

Raises

Exception	Description
SecretNotFoundError	In strict mode, if any secret is not found in the store. In non-strict mode, missing secrets are logged as warnings and boot continues.

Example

from lexigram.security.secrets import EnvSecretStore

app.register_secrets_from_store(
    EnvSecretStore(),
    ["NEXTAUTH_SECRET", "DATABASE_PASSWORD"],
)

from lexigram.security.secrets import EnvSecretStore

app.register_secrets_from_store(
    EnvSecretStore(),
    ["NEXTAUTH_SECRET", "DATABASE_PASSWORD"],
)

add_middleware

def add_middleware(middleware: MiddlewareProtocol[Any, Any]) -> None

source

Add a middleware to the application.

add_provider

def add_provider(provider: Provider) -> None

source

Add a provider to be orchestrated.

Must be called before start().

add_providers

def add_providers(providers: list[Provider]) -> None

source

Add multiple providers. Must be called before start().

discover_providers

def discover_providers(*packages: str) -> None

source

Discover and add Provider subclasses from packages.

Scans each package recursively for Provider subclasses with a no-argument constructor and registers them.

Also discovers @injectable / @singleton decorated classes and schedules them for auto-registration into the container during boot.

Convention: place providers in <module>/provider.py or <module>/providers/ — any file is scanned.

Parameters

Parameter	Type	Description

Example

app.discover_providers("my_platform.modules", "my_platform.infrastructure")

app.discover_providers("my_platform.modules", "my_platform.infrastructure")

add_module

def add_module(module: type | Any) -> None

source

Add a module to the application.

Modules organize providers and define visibility boundaries. Must be called before start().

add_modules

def add_modules(modules: list[type | Any]) -> None

source

Add multiple modules. Must be called before start().

discover_modules

def discover_modules(
    *,
    entry_point_group: str = 'lexigram.modules',
    directories: list[str | Path] | None = None,
    enabled: list[str] | None = None,
    disabled: list[str] | None = None
) -> None

source

Discover Module classes and register them with this application.

Scans importlib.metadata entry points and optionally filesystem directories. Each discovered Module is passed to add_module(), ensuring it flows through CompiledModuleGraph during start().

Parameters

Parameter	Type	Description
`entry_point_group`	str	Entry-point group to scan. Default "lexigram.modules".
`directories`	list[str \| Path] \| None	Extra filesystem directories to scan.
`enabled`	list[str] \| None	Allowlist of module names (empty = all allowed).
`disabled`	list[str] \| None	Denylist of module names to skip.

Example

app.discover_modules()
app.discover_modules(directories=["./plugins"], disabled=["dev-only"])

app.discover_modules()
app.discover_modules(directories=["./plugins"], disabled=["dev-only"])

start

async def start() -> None

source

Boot all providers and start the application.

Phases: 1. Emit ApplicationStarting event 2. Register all providers with the container 3. Boot all providers (dependency order, parallel where possible) 4. Resolve EventBusProtocol from container for lifecycle events 5. Emit ApplicationStarted event

If any phase fails, already-booted providers are shut down and the application moves to STOPPED.

Raises

Exception	Description
RuntimeError	If application is not in CREATED state.

stop

async def stop() -> None

source

Shutdown all providers in reverse order.

Safe to call multiple times. Handles being called from STARTING state (failed boot) as well as RUNNING.

Emits ApplicationStopping before shutdown and ApplicationStopped after completion.

health_check

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

source

Aggregate health check from all providers.

Returns an AggregateHealthResult whose status follows worst-case aggregation across all registered provider health checks.

liveness

async def liveness(timeout: float = 5.0) -> AggregateHealthResult

source

Run liveness checks for the application.

readiness

async def readiness(timeout: float = 5.0) -> AggregateHealthResult

source

Run readiness checks for the application.

startup_check

async def startup_check(timeout: float = 5.0) -> AggregateHealthResult

source

Run startup checks for the application.

boot

async def boot(
    cls,
    name: str = 'lexigram-app',
    providers: list[Provider] | None = None,
    modules: list[Any] | None = None,
    config: LexigramConfig | None = None
) -> AsyncIterator[Application]

source

Create, start, yield, and guarantee shutdown.

Usage

async with Application.boot(providers=[...]) as app:
    await app.invoke(main)

async with Application.boot(providers=[...]) as app:
    await app.invoke(main)

`ConfigProvider`

source

Configuration loading, models, and sources DI provider.

This provider handles loading configuration from various sources (YAML, Env) and registering the configuration instance into the DI container.

All config classes are backed by DomainModel (stdlib dataclasses). Environment variable reading is handled by EnvironmentConfigSource, which is always added as a source.

__init__

def __init__(
    config_class: type[Any] = LexigramConfig,
    sources: list[Any] | None = None,
    env_prefix: str = '',
    extensions: dict[str, type] | None = None
) -> None

source

async def register(container: ContainerRegistrarProtocol) -> None

source

Load configuration and register as a singleton.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

No boot-time work required for the config module.

shutdown

async def shutdown() -> None

source

No resources to release for the config module.

health_check

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

source

Report config loading status.

`ConfigurationError`

source

Configuration loading, parsing, or validation failed.

Raised by:

ConfigProvider during initialisation
Config loaders when sources are invalid
Validation of ConfigProtocol implementations

`Container`

source

Dependency injection container.

Manages service registration, resolution, and lifecycle. Delegates to ContainerRegistrarImpl, ContainerResolverImpl, and ContainerValidator for write, read, and validation responsibilities respectively.

Usage

container = Container()
container.singleton(MyService, MyService())
container.transient(IRepo, SqlRepo)

service = await container.resolve(MyService)

container = Container()
container.singleton(MyService, MyService())
container.transient(IRepo, SqlRepo)

service = await container.resolve(MyService)

__init__

def __init__(
    parent: Container | None = None,
    testing_mode: bool = False
) -> None

source

registrar

property registrar() -> ContainerRegistrarImpl

source

Return the write-only registrar component.

resolver

property resolver() -> ContainerResolverImpl

source

Return the read-only resolver component.

create_invoker

def create_invoker() -> FunctionInvoker

source

Create a FunctionInvoker bound to this container.

Returns a new invoker that shares this container’s service resolver and type-hint cache, so repeated calls to FunctionInvoker.call benefit from the warm cache without exposing the resolver’s private attributes.

Returns

Type	Description
FunctionInvoker	A FunctionInvoker configured with this container's resolver and hints cache.

Example

invoker = container.create_invoker()
result = await invoker.call(my_function)

invoker = container.create_invoker()
result = await invoker.call(my_function)

clear

def clear() -> None

source

Clear all registered services.

Raises

Exception	Description
ContainerError	if the container has been frozen.

transient

def transient(
    service_type: type[T],
    factory: ServiceFactory[T],
    validate: bool = True
) -> None

source

singleton

def singleton(
    service_type: type[T],
    instance: T | None = None,
    *,
    name: str | None = None,
    factory: ServiceFactory[T] | None = None,
    validate: bool = True
) -> None

source

singleton

def singleton(
    service_type: Any,
    instance: Any = None,
    *,
    name: str | None = None,
    factory: Any | None = None,
    validate: bool = True
) -> None

source

singleton

def singleton(
    service_type: Any,
    instance: Any = None,
    *,
    name: str | None = None,
    factory: Any | None = None,
    validate: bool = True
) -> None

source

Pass instance for a pre-built singleton, or factory for lazy creation. Use name for named registrations resolvable via Annotated[T, Named('...')]].

scoped

def scoped(
    service_type: type[T],
    factory: ServiceFactory[T],
    validate: bool = True,
    *,
    name: str | None = None
) -> None

source

override

def override(
    service_type: type[T],
    instance: T
) -> None

source

Replace a service registration for testing purposes.

This method is restricted to containers created with testing_mode=True. It works even on frozen containers, intended for test scenarios where you need to swap a real service with a fake or mock.

Parameters

Parameter	Type	Description
`service_type`	type[T]	The service type to override.
`instance`	T	The replacement instance.

Raises

Exception	Description
ContainerError	If the container is not in testing mode.
ContainerError	If the service is not registered.

is_frozen

property is_frozen() -> bool

source

Return True if this container has been frozen.

Once frozen, no further service registrations are allowed. Use this in tests or startup checks to verify container state.

freeze

def freeze(validate: bool = True) -> None

source

Prevent any further service registrations on this container.

After freezing, calls to transient(), singleton(), scoped() or clear() will raise ContainerError. Freezing is intended to be invoked once the registration phase is complete (e.g. after ProviderOrchestrator.register_all()).

Parameters

Parameter	Type	Description
`validate`	bool	When ``True`` (default), runs pre-flight dependency validation before freezing. Raises ContainerValidationError if any missing dependencies, circular references, or scope violations are detected.

Raises

Exception	Description
ContainerValidationError	If ``validate=True`` and validation issues are found.

resolve

async def resolve(
    service_type: type[T],
    *,
    bypass_visibility: bool = False
) -> T

source

resolve

async def resolve(
    service_type: Any,
    *,
    bypass_visibility: bool = False
) -> Any

source

resolve

async def resolve(
    service_type: Any,
    *,
    bypass_visibility: bool = False
) -> Any

source

Asynchronously resolve a service by its registered type.

Parameters

Parameter	Type	Description
`service_type`	Any	The service type to resolve.
`bypass_visibility`	bool	If True, skip module visibility enforcement. Use only in framework-internal resolution paths.

Returns

Type	Description
Any	The resolved service instance.

Raises

Exception	Description
ModuleVisibilityError	If the current module context does not have visibility over the requested service type.

resolve_sync

def resolve_sync(service_type: type[T]) -> T

source

Synchronously resolve an already-instantiated singleton.

This is intended for use in decorators or other synchronous contexts where async resolution is not possible. It ONLY works for singletons that have already been instantiated (e.g. during boot).

Parameters

Parameter	Type	Description
`service_type`	type[T]	The service type to resolve.

Returns

Type	Description
T	The resolved instance.

Raises

Exception	Description
UnresolvableDependencyError	If not a singleton or not instantiated.

validate

def validate() -> list[str]

source

Validate the container configuration.

Checks:

All registered services have resolvable dependencies (missing dependencies)
No circular dependencies across the entire graph
No scope violations (singleton depending on scoped/transient)

Returns

Type	Description
list[str]	List of validation issues (empty if valid).

validate_no_orphans

def validate_no_orphans() -> list[OrphanedRegistration]

source

Find registrations that no other service depends on.

This is a development-time validator to identify dead code services that are registered but never used.

Returns

Type	Description
list[OrphanedRegistration]	List of potentially orphaned registrations.

has

def has(service_type: object) -> bool

source

Check if a service is explicitly registered.

resolve_optional

async def resolve_optional(service_type: type[T]) -> T | None

source

resolve_optional

async def resolve_optional(service_type: Any) -> Any | None

source

resolve_optional

async def resolve_optional(service_type: Any) -> Any | None

source

Resolve a service, returning None if not registered.

Provides graceful handling for optional dependencies without requiring callers to catch exceptions.

Parameters

Parameter	Type	Description
`service_type`	Any	The type to resolve.

Returns

Type	Description
Any \| None	The resolved instance or None if the service is not registered.

resolve_all

async def resolve_all(service_type: type[T]) -> list[T]

source

resolve_all

async def resolve_all(service_type: Any) -> list[Any]

source

resolve_all

async def resolve_all(service_type: Any) -> list[Any]

source

Resolve all registered implementations that are subtypes of a service type.

Parameters

Parameter	Type	Description
`service_type`	Any	The base type whose implementations to resolve.

Returns

Type	Description
list[Any]	A list of resolved instances for matching registrations.

is_singleton

def is_singleton(service_type: object) -> bool

source

Check if a service is registered as a singleton.

dump_registrations

def dump_registrations() -> list[dict[str, Any]]

source

Return a JSON-serialisable snapshot of all container registrations.

dump_dependency_graph

def dump_dependency_graph() -> dict[str, list[str]]

source

Return an adjacency map of service → direct dependency names.

log_registrations

def log_registrations() -> None

source

Log a human-readable table of all container registrations.

create_scope

def create_scope() -> Scope

source

Create a request-scoped resolution context.

scope

async def scope() -> AsyncIterator[Scope]

source

Async context manager for a request-scoped container.

Creates a new Scope and disposes it automatically when the context exits.

Usage

async with container.scope() as scoped:
    service = await scoped.resolve(UserService)

async with container.scope() as scoped:
    service = await scoped.resolve(UserService)

Yields

Type	Description
AsyncIterator[Scope]	A fresh Scope instance.

call

async def call(
    func: Callable[Ellipsis, Awaitable[T] | T],
    *args: Any,
    **kwargs: Any
) -> T

source

Call a function with dependency injection (Protocol implementation).

dispose

async def dispose() -> None

source

Dispose the container and all internal singletons.

Iterates through all registered singletons and disposes of those that implement AsyncDisposableProtocol or have a close/aclose method.

`ContainerBuilder`

source

Fluent API for building DI containers.

This builder provides a clean, declarative way to configure services and their dependencies for testing and programmatic container construction.

For application-level composition, use Application instead — it provides the full lifecycle (register → freeze → validate → boot → shutdown) and module compilation.

Example

container = await (
    ContainerBuilder()
    .add_singleton(DatabaseConfig, lambda: db_config)
    .add_scoped(UserRepository, SQLAlchemyUserRepository)
    .add_transient(UserService, UserService)
    .build()
)

container = await (
    ContainerBuilder()
    .add_singleton(DatabaseConfig, lambda: db_config)
    .add_scoped(UserRepository, SQLAlchemyUserRepository)
    .add_transient(UserService, UserService)
    .build()
)

__init__

def __init__() -> None

source

Initialize the container builder.

add_singleton

def add_singleton(
    interface: type[T],
    implementation: type[T] | Callable[[], T],
    metadata: dict[str, Any] | None = None
) -> Self

source

Parameters

Parameter	Type	Description
`interface`	type[T]	The service interface/contract.
`implementation`	type[T] \| Callable[[], T]	The concrete class or factory function.
`metadata`	dict[str, Any] \| None	Optional metadata.

Returns

Type	Description
Self	Self for method chaining.

add_instance

def add_instance(
    interface: type[T],
    instance: T
) -> Self

source

Parameters

Parameter	Type	Description
`interface`	type[T]	The service interface/contract.
`instance`	T	The pre-built instance.

Returns

Type	Description
Self	Self for method chaining.

add_scoped

def add_scoped(
    interface: type[T],
    implementation: type[T] | Callable[Ellipsis, T],
    metadata: dict[str, Any] | None = None
) -> Self

source

Parameters

Parameter	Type	Description
`interface`	type[T]	The service interface/contract.
`implementation`	type[T] \| Callable[Ellipsis, T]	The concrete class or factory function.
`metadata`	dict[str, Any] \| None	Optional metadata.

Returns

Type	Description
Self	Self for method chaining.

add_transient

def add_transient(
    interface: type[T],
    implementation: type[T] | Callable[Ellipsis, T],
    metadata: dict[str, Any] | None = None
) -> Self

source

Parameters

Parameter	Type	Description
`interface`	type[T]	The service interface/contract.
`implementation`	type[T] \| Callable[Ellipsis, T]	The concrete class or factory function.
`metadata`	dict[str, Any] \| None	Optional metadata.

Returns

Type	Description
Self	Self for method chaining.

add_module

def add_module(module_or_dynamic: type | DynamicModule) -> Self

source

Add a module to configure the container.

Accepts module classes (decorated with @module) and DynamicModule instances.

Module providers are registered during build. The builder performs basic module processing (provider extraction) but does NOT run the full ModuleCompiler — it does not validate import/export visibility. For full module validation, use Application.

Parameters

Parameter	Type	Description
`module_or_dynamic`	type \| DynamicModule	A module class or DynamicModule.

Returns

Type	Description
Self	Self for method chaining.

add_provider

def add_provider(provider: Any) -> Self

source

Add a provider whose register() will be called during build.

Parameters

Parameter	Type	Description
`provider`	Any	A Provider instance.

Returns

Type	Description
Self	Self for method chaining.

disable_validation

def disable_validation() -> Self

source

Disable container validation during build.

Returns

Type	Description
Self	Self for method chaining.

build

async def build() -> ContainerResolverProtocol

source

Build and validate the container.

Processing order: 1. Extract providers from modules 2. Register module providers 3. Apply direct service registrations 4. Call provider.register() on standalone providers 5. Validate (if enabled)

Returns

Type	Description
ContainerResolverProtocol	A ContainerResolverProtocol for resolving services.

Raises

Exception	Description
ContainerBuildError	If build or validation fails.

`Context`

source

User-facing typed wrapper around an injected ``ContextVarRegistry``.

__init__

def __init__(registry: ContextVarRegistry) -> None

source

registry

property registry() -> ContextVarRegistry

source

The underlying registry (for advanced use / wiring).

get

def get(
    key: ContextKey[T],
    default: T | None = None
) -> T | None

source

Get the current value for a typed key.

set

def set(
    key: ContextKey[T],
    value: T
) -> contextvars.Token[T | None]

source

Set a typed value, returning a reset token.

reset

def reset(
    key: ContextKey[T],
    token: contextvars.Token[T | None]
) -> None

source

Reset a typed value using a token from set.

register_key

def register_key(key: ContextKey[Any]) -> None

source

get_dynamic

def get_dynamic(
    key: str,
    default: Any = None
) -> Any

source

Get a value by string key.

set_dynamic

def set_dynamic(
    key: str,
    value: Any
) -> contextvars.Token[Any]

source

Set a value by string key.

reset_dynamic

def reset_dynamic(
    key: str,
    token: contextvars.Token[Any]
) -> None

source

Reset a value by string key.

snapshot

def snapshot() -> dict[str, Any]

source

Snapshot of all non-None context values.

has

def has(key: str) -> bool

source

Check whether a context key is registered.

`ContextKey`

source

Immutable, typed key for context-variable access (pure data).

`CoreInfrastructureProvider`

source

Provider for **shared low-level Lexigram infrastructure**.

This is a leaf provider (no sub-providers) that registers the primitive cross-cutting services needed by every other provider:

Context / ContextVarRegistry — request-scoped context propagation
Registry — a named-object registry for ad-hoc service look-up

Do NOT confuse with CoreProvider:

+-----------------------------------+----------------------------------------------------+ | CoreInfrastructureProvider | CoreProvider | +===================================+====================================================+ | Leaf provider, no sub-providers | Aggregating provider, no direct registrations | +-----------------------------------+----------------------------------------------------+ | Registers concrete primitives | Delegates to sub-providers via orchestrator | +-----------------------------------+----------------------------------------------------+ | Used inside the framework | Used by application bootstrap (LexigramApp) | +-----------------------------------+----------------------------------------------------+ | name = "common" | name = "core" | +-----------------------------------+----------------------------------------------------+

async def register(container: ContainerRegistrarProtocol) -> None

source

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Initialize common services.

shutdown

async def shutdown() -> None

source

Clean up common resources.

health_check

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

source

Check health of shared infrastructure.

`CoreModule`

source

Kernel provider manifest for a minimal Lexigram application.

Bundles the four kernel framework providers that every Lexigram application requires: configuration, logging, infrastructure primitives, and DI introspection.

For a batteries-included bundle that also provides serialization and concurrency, use StandardModule.

Extension packages add their own providers separately. CoreModule has zero cross-package imports — it references only providers defined within the lexigram core package and contracts from lexigram-contracts.

build_providers

def build_providers(
    cls,
    config_class: type[Any] | None = None,
    config_sources: list[Any] | None = None,
    env_prefix: str = '',
    overrides: dict[str, Provider] | None = None
) -> list[Provider]

source

Instantiate and return the core provider list.

Creates a fresh provider instance from each registered core provider class. Use overrides to replace any default provider by its name.

Parameters

Parameter	Type	Description
`config_class`	type[Any] \| None	Custom configuration dataclass to use instead of the default LexigramConfig.
`config_sources`	list[Any] \| None	Explicit list of ConfigSource instances. When ``None``, ``ConfigProvider`` falls back to its default source chain (environment variables + config files).
`env_prefix`	str	Environment variable prefix forwarded to the ``ConfigProvider``.
`overrides`	dict[str, Provider] \| None	Mapping of provider name → replacement Provider instance. A name matching a core provider replaces that provider; unmatched names are appended to the end of the list.

Returns

Type	Description
list[Provider]	Ordered list of Provider instances ready to pass to Application.boot.

Example

providers = CoreModule.build_providers(
    config_class=MyConfig,
    overrides={"logging": CustomLoggingProvider()},
)

providers = CoreModule.build_providers(
    config_class=MyConfig,
    overrides={"logging": CustomLoggingProvider()},
)

configure

def configure(
    cls,
    config_class: type[Any] | None = None,
    config_sources: list[Any] | None = None,
    env_prefix: str = '',
    overrides: dict[str, Provider] | None = None
) -> DynamicModule

source

Create a DynamicModule configured with core provider instances.

Parameters

Parameter	Type	Description
`config_class`	type[Any] \| None	Custom configuration dataclass for ``ConfigProvider``.
`config_sources`	list[Any] \| None	Explicit config source list for ``ConfigProvider``.
`env_prefix`	str	Environment variable prefix for ``ConfigProvider``.
`overrides`	dict[str, Provider] \| None	Mapping of provider name -> replacement provider instance.

Returns

Type	Description
DynamicModule	DynamicModule configured with instantiated core providers and default core exports.

`CoreProvider`

source

Main **aggregating** provider for Lexigram core services.

CoreProvider is the single entry point for application bootstrap. It does not register services directly — instead it relies on the orchestrator to discover and execute all framework sub-providers in dependency order.

Do NOT confuse with CoreInfrastructureProvider:

+-----------------------------------+----------------------------------------------------+ | CoreProvider | CoreInfrastructureProvider | +===================================+====================================================+ | Aggregating, no registrations | Leaf provider, registers concrete primitives | +-----------------------------------+----------------------------------------------------+ | Used by application bootstrap | Used inside the framework internals | +-----------------------------------+----------------------------------------------------+ | name = "core" | name = "common" | +-----------------------------------+----------------------------------------------------+

__init__

def __init__() -> None

source

Initialize CoreProvider.

async def register(container: ContainerRegistrarProtocol) -> None

source

Note: Sub-providers are now handled by the orchestrator via the sub_providers property and recursive discovery.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Boot core services.

shutdown

async def shutdown() -> None

source

Shutdown core services.

health_check

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

source

CoreProvider itself is always healthy if it reached boot.

Individual core services (config, logging, etc.) are checked by their own providers via the orchestrator.

`DomainModelError`

source

Domain model invariant violated or business rule broken.

Raised by:

Domain event handlers
Aggregate root validations
Entity lifecycle violations

`DynamicModule`

source

Runtime-configured module descriptor.

Returned by factory methods like Module.configure() and Module.scope(). When a DynamicModule appears in an import list or is passed to Application.add_module(), it fully replaces the static @module() metadata for the referenced module class.

Provider entries may be classes (instantiated by the orchestrator) or pre-constructed instances (registered directly)

DynamicModule(
    module=CacheModule,
    providers=[
        CacheProvider(backend="redis"),   # instance
        MetricsProvider,                  # class
    ],
    exports=[CacheBackendProtocol],
    is_global=True,
)

DynamicModule(
    module=CacheModule,
    providers=[
        CacheProvider(backend="redis"),   # instance
        MetricsProvider,                  # class
    ],
    exports=[CacheBackendProtocol],
    is_global=True,
)

Identity and Deduplication: The module field determines identity. Two DynamicModule entries referencing the same module class but with different configurations raise ModuleDuplicateError at compile time.

resolved_name

property resolved_name() -> str

source

Module name: explicit name or module.__name__.

`Err`

source

__init__

def __init__(error: E) -> None

source

is_ok

def is_ok() -> bool

source

is_err

def is_err() -> bool

source

unwrap

def unwrap() -> T

source

unwrap_err

def unwrap_err() -> E

source

unwrap_or

def unwrap_or(default: T) -> T

source

unwrap_or_else

def unwrap_or_else(op: Callable[[E], T]) -> T

source

map_sync

def map_sync(op: Callable[[T], U]) -> Result[U, E]

source

map_err

def map_err(op: Callable[[E], F]) -> Result[T, F]

source

and_then_sync

def and_then_sync(op: Callable[[T], Result[U, E]]) -> Result[U, E]

source

or_else_sync

def or_else_sync(op: Callable[[E], Result[T, F]]) -> Result[T, F]

source

expect

def expect(message: str) -> T

source

match

def match(
    ok: Callable[[T], U],
    err: Callable[[E], U]
) -> U

source

map

async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

source

async_map

async def async_map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

source

Alias for map — exists for backward compatibility.

and_then

async def and_then(op: Callable[[T], Awaitable[Result[U, E]]]) -> Result[U, E]

source

or_else

async def or_else(op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]

source

flatten

def flatten() -> Result[Any, E]

source

filter

def filter(
    predicate: Callable[[T], bool],
    error: E
) -> Result[T, E]

source

ok_or

def ok_or(default: U) -> U

source

`HookPriority`

source

Named priority levels for hook handlers.

Lower values run first. Custom integer values are accepted anywhere a priority is expected — these named levels are conveniences.

Spacing between levels leaves room for application-specific intermediate values (e.g. HookPriority.EARLY + 10).

`HookRegistry`

source

Named hook registry for framework extensibility.

Thread-safe registry for action hooks (fire-and-forget, error-isolated) and filter hooks (value-pipeline, error-propagating) with priority ordering.

Read operations (has_*, call_action, apply_filter) are lock-free — they snapshot the handler list and iterate the snapshot. Write operations (register_*, unregister_*) acquire the lock.

Parameters

Parameter	Type	Description
`name`		Registry name used in structured log output. Defaults to ``"default"``.

Example

hooks = HookRegistry("app")

@hooks.action("before_request", priority=HookPriority.EARLY)
async def log_request(**kwargs: Any) -> None:
    logger.info("incoming", path=kwargs["path"])

await hooks.call_action("before_request", path="/api/users")

hooks = HookRegistry("app")

@hooks.action("before_request", priority=HookPriority.EARLY)
async def log_request(**kwargs: Any) -> None:
    logger.info("incoming", path=kwargs["path"])

await hooks.call_action("before_request", path="/api/users")

__init__

def __init__(name: str = 'default') -> None

source

register_action

def register_action(
    hook_name: str,
    handler: Callable[Ellipsis, Any],
    priority: int = HookPriority.NORMAL,
    *,
    once: bool = False
) -> None

source

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.
`handler`	Callable[Ellipsis, Any]	Callable to invoke. Can be sync or async.
`priority`	int	Lower values run first. Defaults to ``HookPriority.NORMAL`` (100).
`once`	bool	If True, auto-remove handler after first invocation.

register_filter

def register_filter(
    hook_name: str,
    handler: Callable[Ellipsis, Any],
    priority: int = HookPriority.NORMAL,
    *,
    once: bool = False
) -> None

source

Filter handlers receive a value as the first positional argument and must return the (possibly transformed) value.

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.
`handler`	Callable[Ellipsis, Any]	Callable that receives and returns a value.
`priority`	int	Lower values run first. Defaults to ``HookPriority.NORMAL`` (100).
`once`	bool	If True, auto-remove handler after first invocation.

unregister_action

def unregister_action(
    hook_name: str,
    handler: Callable[Ellipsis, Any]
) -> bool

source

Remove an action handler from a hook.

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.
`handler`	Callable[Ellipsis, Any]	The handler to remove (identity comparison).

Returns

Type	Description
bool	True if the handler was found and removed.

unregister_filter

def unregister_filter(
    hook_name: str,
    handler: Callable[Ellipsis, Any]
) -> bool

source

Remove a filter handler from a hook.

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.
`handler`	Callable[Ellipsis, Any]	The handler to remove (identity comparison).

Returns

Type	Description
bool	True if the handler was found and removed.

call_action

async def call_action(
    hook_name: str,
    **kwargs: Any
) -> None

source

Invoke all action handlers for the named hook.

Handlers run sequentially in priority order. Errors in individual handlers are logged but do not prevent subsequent handlers from running.

Once-handlers are removed after they fire (even if they raise).

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name. **kwargs: Arguments passed to each handler.

apply_filter

async def apply_filter(
    hook_name: str,
    value: Any,
    **kwargs: Any
) -> Any

source

Pass a value through all filter handlers for the named hook.

Each handler receives the value (possibly transformed by prior handlers) and must return the new value. Handlers run sequentially in priority order.

Once-handlers are removed after they fire (even if they raise). Errors abort the pipeline and re-raise to the caller.

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.
`value`	Any	The initial value to filter. **kwargs: Additional arguments passed to each handler.

Returns

Type	Description
Any	The final filtered value.

has_action

def has_action(hook_name: str) -> bool

source

Check if any action handlers are registered for a hook.

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.

Returns

Type	Description
bool	True if at least one handler is registered.

has_filter

def has_filter(hook_name: str) -> bool

source

Check if any filter handlers are registered for a hook.

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.

Returns

Type	Description
bool	True if at least one handler is registered.

action_names

def action_names() -> list[str]

source

List all registered action hook names.

Returns

Type	Description
list[str]	List of hook names that have at least one action handler.

filter_names

def filter_names() -> list[str]

source

List all registered filter hook names.

Returns

Type	Description
list[str]	List of hook names that have at least one filter handler.

clear

def clear(hook_name: str | None = None) -> None

source

Clear all handlers, or handlers for a specific hook.

Parameters

Parameter	Type	Description
`hook_name`	str \| None	If provided, clear only this hook. Otherwise clear all.

action

def action(
    hook_name: str,
    *,
    priority: int = HookPriority.NORMAL,
    once: bool = False
) -> Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]

source

Decorator to register an action handler.

Returns the handler unchanged so it can still be called directly.

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.
`priority`	int	Lower values run first. Defaults to ``HookPriority.NORMAL``.
`once`	bool	If True, auto-remove handler after first invocation.

Returns

Type	Description
Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]	Decorator that registers the handler and returns it unchanged.

Example

@hooks.action("before_request", priority=HookPriority.EARLY)
async def log_request(**kwargs: Any) -> None:
    logger.info("request", path=kwargs["path"])

@hooks.action("before_request", priority=HookPriority.EARLY)
async def log_request(**kwargs: Any) -> None:
    logger.info("request", path=kwargs["path"])

filter

def filter(
    hook_name: str,
    *,
    priority: int = HookPriority.NORMAL,
    once: bool = False
) -> Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]

source

Decorator to register a filter handler.

Returns the handler unchanged so it can still be called directly.

Parameters

Parameter	Type	Description
`hook_name`	str	The hook point name.
`priority`	int	Lower values run first. Defaults to ``HookPriority.NORMAL``.
`once`	bool	If True, auto-remove handler after first invocation.

Returns

Type	Description
Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]	Decorator that registers the handler and returns it unchanged.

Example

@hooks.filter("transform_response")
async def add_headers(value: Any, **kwargs: Any) -> Any:
    value.headers["X-Custom"] = "yes"
    return value

@hooks.filter("transform_response")
async def add_headers(value: Any, **kwargs: Any) -> Any:
    value.headers["X-Custom"] = "yes"
    return value

`IdentityProvider`

source

__init__

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

source

async def register(container: ContainerRegistrarProtocol) -> None

source

boot

async def boot(container: BootContainerProtocol) -> None

source

Install the resolved generator into ambient.

shutdown

async def shutdown() -> None

source

Identity services do not require shutdown work.

health_check

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

source

Report the identity subsystem as healthy.

`Injectable`

source

Decorator class to mark a class as injectable for dependency injection.

This decorator marks classes for automatic registration with the dependency injection container. The container can then resolve these services and automatically inject their dependencies.

Parameters

Parameter	Type	Description
`scope`		The service scope — transient (new instance each time), singleton (shared instance), or scoped (instance per scope). Defaults to transient.
`name`		Optional name for the service registration. If not provided, the class name is used.

__init__

def __init__(
    scope: ServiceScope = ServiceScope.TRANSIENT,
    name: str | None = None
) -> None

source

Initialize the Injectable decorator.

Parameters

Parameter	Type	Description
`scope`	ServiceScope	The service scope for this class.
`name`	str \| None	Optional name for the service.

`InjectionError`

source

Dependency injection or container resolution failed.

Raised by:

Container.resolve() when a service is not registered
Provider.register() on configuration errors
Module resolution when imports fail
Circular dependency detection

`Invoker`

source

Invokes functions with dependency injection and middleware.

This class decouples the invocation logic from the Application class, allowing it to be used in other contexts (like testing or custom runners).

The context dict passed through the middleware pipeline satisfies InvocationContextProtocol — a transport-neutral invocation context that transports can extend with typed attributes in Phase 3.

__init__

def __init__(
    container: Container,
    middleware: MiddlewarePipelineProtocol
) -> None

source

invoke

async def invoke(
    func: Callable[Ellipsis, Awaitable[T]],
    *args: Any,
    **kwargs: Any
) -> T

source

Invoke a function with automatic dependency injection and middleware.

The invocation context passed through the middleware pipeline is a dict[str, Any] that structurally satisfies InvocationContextProtocol. Middleware implementations should treat it as opaque and forward it unchanged unless they are deliberately enriching the context.

`LoggingProvider`

source

Structured logging factory and helpers DI provider.

async def register(container: ContainerRegistrarProtocol) -> None

source

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Apply logging configuration from LexigramConfig.

shutdown

async def shutdown() -> None

source

No resources to release for the logging module.

health_check

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

source

Health check — always healthy (in-process only, no external backend).

Parameters

Parameter	Type	Description
`timeout`	float	Ignored for in-process providers.

Returns

Type	Description
HealthCheckResult	Always HEALTHY — no external backend to check.

`MiddlewareChain`

source

Mutable builder for MiddlewarePipeline.

Collect middleware via add (fluent interface supported), then call build to produce the frozen MiddlewarePipeline that is safe to share.

The chain itself is never executed — it is only a construction helper. After build is called the chain remains mutable; further add calls extend the next build.

__init__

def __init__(middleware: list[Any] | None = None) -> None

source

Create a chain, optionally pre-populated with middleware.

Parameters

Parameter	Type	Description
`middleware`	list[Any] \| None	Initial list of middleware to seed the chain with.

add

def add(middleware: Any) -> MiddlewareChain

source

Append a middleware and return self for chaining.

Parameters

Parameter	Type	Description
`middleware`	Any	An ``async (context, next_handler) -> result`` callable.

Returns

Type	Description
MiddlewareChain	``self`` so calls can be chained fluently.

build

def build() -> MiddlewarePipeline

source

Produce an immutable MiddlewarePipeline from the current state.

The returned pipeline is a snapshot — subsequent add calls do not affect already-built pipelines.

Returns

Type	Description
MiddlewarePipeline	A new MiddlewarePipeline containing all added middleware.

clear

def clear() -> None

source

Remove all middleware from the chain without building.

`MiddlewareConfig`

source

Middleware chain, pipeline, and registry configuration.

`MiddlewareModule`

source

Composable async middleware pipeline, exception filter chain, and request pipeline.

Call configure to configure the middleware subsystem with custom settings.

Usage

from lexigram.middleware.config import MiddlewareConfig

@module(
    imports=[MiddlewareModule.configure(MiddlewareConfig(...))]
)
class AppModule(Module):
    pass

from lexigram.middleware.config import MiddlewareConfig

@module(
    imports=[MiddlewareModule.configure(MiddlewareConfig(...))]
)
class AppModule(Module):
    pass

configure

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

source

Create a MiddlewareModule with explicit configuration.

Parameters

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

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

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

source

Return a MiddlewareModule with framework defaults for testing.

All middleware uses default configuration. No request pipeline is active — only the DI bindings are registered.

Parameters

Parameter	Type	Description
`config`	Any \| None	Accepted for interface compatibility with stub; ignored. Pass MiddlewareConfig to configure instead.

Returns

Type	Description
DynamicModule	A DynamicModule with default middleware configuration.

`MiddlewareProvider`

source

Middleware chain, pipeline, and registry DI provider.

On register():

Registers MiddlewareChain, ExceptionFilterChain, and MiddlewarePipeline as container singletons.
Scans the lexigram.middleware entry-point group and delegates registration to every discovered Provider subclass so that third-party middleware packages integrate automatically.

__init__

def __init__() -> None

source

async def register(container: ContainerRegistrarProtocol) -> None

source

Parameters

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

boot

async def boot(container: BootContainerProtocol) -> None

source

Boot discovered providers after middleware chains are initialized.

Parameters

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

shutdown

async def shutdown() -> None

source

Shut down discovered providers in reverse order.

Providers are shut down in reverse discovery order to respect dependency cleanup semantics (last booted = first shut down).

health_check

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

source

Check provider health.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for health check response.

Returns

Type	Description
HealthCheckResult	HealthCheckResult with status and component details.

`MiddlewareRegistry`

source

Named middleware registry with priority ordering.

Inherits from Registry with priority support for ordering middleware by priority. Middleware is stored by name and retrieved in ascending priority order (lower number = earlier execution).

The registry is thread-safe through the base Registry implementation.

Example

registry = MiddlewareRegistry()
registry.register("auth", AuthMiddleware(), priority=10)
registry.register("logging", LoggingMiddleware(), priority=20)

# Returns middleware in priority order (auth first, then logging)
middleware_list = registry.all()

registry = MiddlewareRegistry()
registry.register("auth", AuthMiddleware(), priority=10)
registry.register("logging", LoggingMiddleware(), priority=20)

# Returns middleware in priority order (auth first, then logging)
middleware_list = registry.all()

__init__

def __init__() -> None

source

Initialize the middleware registry with priority ordering.

def register(
    name: str,
    middleware: Any,
    *,
    priority: int = 50
) -> None

source

Parameters

Parameter	Type	Description
`name`	str	Unique name for this middleware.
`middleware`	Any	The middleware instance.
`priority`	int	Execution order - lower values run first. Default 50.

Raises

Exception	Description
ValueError	If a middleware with the same name is already registered.

unregister

def unregister(name: str) -> Any

source

Remove and return a named middleware.

Parameters

Parameter	Type	Description
`name`	str	The middleware name to remove.

Returns

Type	Description
Any	The removed middleware instance.

Raises

Exception	Description
KeyError	If no middleware with the given name is registered.

get

def get(name: str) -> Any

source

Retrieve a named middleware.

Parameters

Parameter	Type	Description
`name`	str	The middleware name to look up.

Returns

Type	Description
Any	The middleware instance.

Raises

Exception	Description
KeyError	If no middleware with the given name is registered.

all

def all() -> list[Any]

source

Return all middleware in priority order.

Returns

Type	Description
list[Any]	List of middleware instances sorted by ascending priority.

has

def has(name: str) -> bool

source

Check if a middleware with the given name is registered.

names

def names() -> list[str]

source

Return a sorted list of registered middleware names.

build_chain

def build_chain() -> MiddlewareChain

source

Build a MiddlewareChain from all registered middleware.

Returns middleware in ascending priority order (lower priority number executes first).

Returns

Type	Description
MiddlewareChain	A new MiddlewareChain containing all registered middleware.

Example

registry = MiddlewareRegistry()
registry.register("auth", AuthMiddleware(), priority=10)
registry.register("logging", LoggingMiddleware(), priority=20)
chain = registry.build_chain()
result = await chain.execute(request)

registry = MiddlewareRegistry()
registry.register("auth", AuthMiddleware(), priority=10)
registry.register("logging", LoggingMiddleware(), priority=20)
chain = registry.build_chain()
result = await chain.execute(request)

with_defaults

def with_defaults(cls) -> MiddlewareRegistry

source

Pre-populate with standard infrastructure middleware.

`Module`

source

Optional base class for module classes.

Provides ClassVar defaults that the module decorator reads, IDE autocompletion, and the configure() / scope() / stub() factory pattern for creating DynamicModule descriptors.

Inheriting from Module is NOT required — any class decorated with @module() is a valid Lexigram module. This base class is a convenience, not a mandate.

Module instances can also be used directly as class decorators, which enables the root-module composition style familiar from other frameworks:

.. code-block:: python

@Module(imports=[UserModule, ProductModule])
class AppModule(ModuleBase):
    pass

ClassVar Inheritance

class BaseInfra(Module):
    providers = [LoggingProvider]

@module()
class FullInfra(BaseInfra):
    exports = [LoggerProtocol]

class BaseInfra(Module):
    providers = [LoggingProvider]

@module()
class FullInfra(BaseInfra):
    exports = [LoggerProtocol]

Factory Pattern

@module()
class DatabaseModule(Module):

    @classmethod
    def configure(cls, url: str) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[DatabaseProvider(url=url)],
            exports=[DatabaseSession, TransactionManager],
            is_global=True,
        )

    @classmethod
    def scope(cls, *models: type) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[ModelRegistryProvider(models)],
        )

@module()
class DatabaseModule(Module):

    @classmethod
    def configure(cls, url: str) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[DatabaseProvider(url=url)],
            exports=[DatabaseSession, TransactionManager],
            is_global=True,
        )

    @classmethod
    def scope(cls, *models: type) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[ModelRegistryProvider(models)],
        )

__init__

def __init__(
    name: str | None = None,
    providers: list[type] | None = None,
    imports: list[Any] | None = None,
    exports: list[type] | None = None,
    controllers: list[type] | None = None,
    is_global: bool = False,
    scan: list[str] | None = None
) -> None

source

Optionally override ClassVar defaults with instance values.

Parameters

Parameter	Type	Description
`name`	str \| None	Module name.
`providers`	list[type] \| None	Provider classes declared by this module.
`imports`	list[Any] \| None	Module classes or DynamicModule instances to import.
`exports`	list[type] \| None	Contract types to expose to importing modules.
`controllers`	list[type] \| None	Controller classes registered with the web layer.
`is_global`	bool	When ``True``, exports are visible to all modules.
`scan`	list[str] \| None	Package paths to scan for ``@injectable`` / ``@singleton`` classes.

configure

def configure(
    cls,
    *args: Any,
    **kwargs: Any
) -> DynamicModule

source

Configure this module globally. Called once at the app root.

Parameters

Parameter	Type	Description

Returns

Type	Description
DynamicModule	DynamicModule for root import.

Raises

Exception	Description
NotImplementedError	Subclasses must override this method.

scope

def scope(
    cls,
    *providers: type,
    **kwargs: Any
) -> DynamicModule

source

Creates an anonymous sub-class token so the compiler treats this as a distinct module (avoids ModuleDuplicateError when both configure() and scope() are imported into the same compiled graph).

Parameters

Parameter	Type	Description

Returns

Type	Description
DynamicModule	DynamicModule with a unique anonymous token as module identity.

stub

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

source

Return a test-mode module with in-memory/noop backends.

Parameters

Parameter	Type	Description
`config`	Any	Optional test configuration override.

Returns

Type	Description
DynamicModule	DynamicModule using test providers.

Raises

Exception	Description
NotImplementedError	Subclasses must override this method.

on_module_booted

async def on_module_booted(cls) -> None

source

Called after all providers in this module have been booted.

Override in subclasses to run post-boot initialization logic. The container is fully available at this point.

on_module_shutdown

async def on_module_shutdown(cls) -> None

source

Called before this module’s providers are shut down.

Override in subclasses to run pre-shutdown cleanup logic.

`ModuleBase`

source

Optional base class for module classes.

Provides ClassVar defaults that the module decorator reads, IDE autocompletion, and the configure() / scope() / stub() factory pattern for creating DynamicModule descriptors.

Inheriting from Module is NOT required — any class decorated with @module() is a valid Lexigram module. This base class is a convenience, not a mandate.

Module instances can also be used directly as class decorators, which enables the root-module composition style familiar from other frameworks:

.. code-block:: python

@Module(imports=[UserModule, ProductModule])
class AppModule(ModuleBase):
    pass

ClassVar Inheritance

class BaseInfra(Module):
    providers = [LoggingProvider]

@module()
class FullInfra(BaseInfra):
    exports = [LoggerProtocol]

class BaseInfra(Module):
    providers = [LoggingProvider]

@module()
class FullInfra(BaseInfra):
    exports = [LoggerProtocol]

Factory Pattern

@module()
class DatabaseModule(Module):

    @classmethod
    def configure(cls, url: str) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[DatabaseProvider(url=url)],
            exports=[DatabaseSession, TransactionManager],
            is_global=True,
        )

    @classmethod
    def scope(cls, *models: type) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[ModelRegistryProvider(models)],
        )

@module()
class DatabaseModule(Module):

    @classmethod
    def configure(cls, url: str) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[DatabaseProvider(url=url)],
            exports=[DatabaseSession, TransactionManager],
            is_global=True,
        )

    @classmethod
    def scope(cls, *models: type) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[ModelRegistryProvider(models)],
        )

__init__

def __init__(
    name: str | None = None,
    providers: list[type] | None = None,
    imports: list[Any] | None = None,
    exports: list[type] | None = None,
    controllers: list[type] | None = None,
    is_global: bool = False,
    scan: list[str] | None = None
) -> None

source

Optionally override ClassVar defaults with instance values.

Parameters

Parameter	Type	Description
`name`	str \| None	Module name.
`providers`	list[type] \| None	Provider classes declared by this module.
`imports`	list[Any] \| None	Module classes or DynamicModule instances to import.
`exports`	list[type] \| None	Contract types to expose to importing modules.
`controllers`	list[type] \| None	Controller classes registered with the web layer.
`is_global`	bool	When ``True``, exports are visible to all modules.
`scan`	list[str] \| None	Package paths to scan for ``@injectable`` / ``@singleton`` classes.

configure

def configure(
    cls,
    *args: Any,
    **kwargs: Any
) -> DynamicModule

source

Configure this module globally. Called once at the app root.

Parameters

Parameter	Type	Description

Returns

Type	Description
DynamicModule	DynamicModule for root import.

Raises

Exception	Description
NotImplementedError	Subclasses must override this method.

scope

def scope(
    cls,
    *providers: type,
    **kwargs: Any
) -> DynamicModule

source

Creates an anonymous sub-class token so the compiler treats this as a distinct module (avoids ModuleDuplicateError when both configure() and scope() are imported into the same compiled graph).

Parameters

Parameter	Type	Description

Returns

Type	Description
DynamicModule	DynamicModule with a unique anonymous token as module identity.

stub

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

source

Return a test-mode module with in-memory/noop backends.

Parameters

Parameter	Type	Description
`config`	Any	Optional test configuration override.

Returns

Type	Description
DynamicModule	DynamicModule using test providers.

Raises

Exception	Description
NotImplementedError	Subclasses must override this method.

on_module_booted

async def on_module_booted(cls) -> None

source

Called after all providers in this module have been booted.

Override in subclasses to run post-boot initialization logic. The container is fully available at this point.

on_module_shutdown

async def on_module_shutdown(cls) -> None

source

Called before this module’s providers are shut down.

Override in subclasses to run pre-shutdown cleanup logic.

`Ok`

source

__init__

def __init__(value: T) -> None

source

is_ok

def is_ok() -> bool

source

is_err

def is_err() -> bool

source

unwrap

def unwrap() -> T

source

unwrap_err

def unwrap_err() -> E

source

unwrap_or

def unwrap_or(default: T) -> T

source

unwrap_or_else

def unwrap_or_else(op: Callable[[E], T]) -> T

source

map_sync

def map_sync(op: Callable[[T], U]) -> Result[U, E]

source

map_err

def map_err(op: Callable[[E], F]) -> Result[T, F]

source

and_then_sync

def and_then_sync(op: Callable[[T], Result[U, E]]) -> Result[U, E]

source

or_else_sync

def or_else_sync(op: Callable[[E], Result[T, F]]) -> Result[T, F]

source

expect

def expect(message: str) -> T

source

match

def match(
    ok: Callable[[T], U],
    err: Callable[[E], U]
) -> U

source

map

async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

source

async_map

async def async_map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

source

Alias for map — exists for backward compatibility.

and_then

async def and_then(op: Callable[[T], Awaitable[Result[U, E]]]) -> Result[U, E]

source

or_else

async def or_else(op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]

source

flatten

def flatten() -> Result[Any, E]

source

filter

def filter(
    predicate: Callable[[T], bool],
    error: E
) -> Result[T, E]

source

ok_or

def ok_or(default: U) -> T

source

`Provider`

source

Base class for all Lexigram providers.

Providers are part of the composition root. They register bindings and wire the object graph during boot.

__init__

def __init__(
    name: str | None = None,
    priority: ProviderPriority | None = None,
    dependencies: tuple[str, Ellipsis] | None = None,
    optional_dependencies: tuple[str, Ellipsis] | None = None,
    boot_timeout: float | None = None,
    required: bool | None = None
) -> None

source

state

property state() -> ProviderState

source

is_booted

property is_booted() -> bool

source

config

property config() -> Any

source

config

def config(value: Any) -> None

source

Allow providers to assign configuration during initialization.

from_config

def from_config(
    cls,
    config: Any,
    **context: Any
) -> Self

source

Create a provider instance from a typed config object.

Subclasses should implement from_config(cls, config: TypedConfig, **context: Any) -> Self with the exact typed config model.

Parameters

Parameter	Type	Description
`config`	Any	Provider-specific configuration object. **context: Optional extra keyword arguments (e.g. pre-built dependencies such as a queue or store instance).

Returns

Type	Description
Self	A new provider instance configured from config.

async def register(container: ContainerRegistrarProtocol) -> None

source

Bind services into the container.

boot

async def boot(container: BootContainerProtocol) -> None

source

Initialize services and wire dependencies.

shutdown

async def shutdown() -> None

source

Tear down resources.

on_error

async def on_error(
    error: Exception,
    phase: str
) -> None

source

Called when boot() or shutdown() raises an exception.

Override to perform cleanup on startup/shutdown failure. This hook allows providers to clean up partial state when lifecycle methods fail.

The default implementation logs the error and does not re-raise.

Parameters

Parameter	Type	Description
`error`	Exception	The exception that was raised.
`phase`	str	Either 'boot' or 'shutdown' indicating which phase failed.

health_check

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

source

Return health status of this provider.

`ProviderPriority`

source

Provider initialization priority.

Providers are booted in ascending order of priority. Lower values run first; the typical ordering is:

CRITICAL (0) – absolutely foundational services that others depend on (e.g. configuration, diagnostics).
INFRASTRUCTURE (10) – low-level plumbing such as database connections, messaging clients, and cache providers.
SECURITY (20) – authentication/authorization infrastructure.
NORMAL (30) – everyday domain services with no special ordering.
APPLICATION (40) – application-level tools (CLI, admin utilities) that depend on domain services but are not themselves domain logic.
DOMAIN (50) – business-logic providers that may depend on earlier layers.
PRESENTATION (80) – web/API layers and other entry points.
COMMS (90) – outbound communication providers (email, SMS, webhooks) often run late to avoid interfering with core initialization.
LOW (100) – lowest-priority, optional providers that can boot last.

`ProviderState`

source

Provider lifecycle state.

`Registry`

source

Unified Registry for managing collections.

Features:

Generic type support
Factory/lazy registration
Lifecycle hooks (on_register, on_unregister)
Decorator support
Thread-safe (using internally managed lock)
Iteration support
Priority ordering support

Registration Patterns

The Registry supports three distinct registration patterns. Choose the one that best fits your use-case:

1. Direct registration — use when the value already exists

registry.register("my_handler", MyHandler())

registry.register("my_handler", MyHandler())

2. Factory (lazy) registration — use when construction is expensive or must be deferred until the first lookup

registry.register_factory("my_handler", lambda: MyHandler())

registry.register_factory("my_handler", lambda: MyHandler())

The factory is called once on the first get() / resolve() and the result is cached. Subsequent lookups return the cached instance.

3. Decorator registration — syntactic sugar for direct registration, useful when defining a class or function inline

@registry.register("my_handler")
class MyHandler:
    ...

@registry.register("my_handler")
class MyHandler:
    ...

Calling registry.register(key) with only the key (no value) activates decorator mode, so the decorated object is registered under key and returned unchanged (enabling chaining with other decorators).

For most framework-internal use adopt the decorator pattern when registering at module load time and the factory pattern when registration must be deferred or is conditional on runtime configuration.

__init__

def __init__(
    name: str | None = None,
    *,
    allow_overwrite: bool = False,
    priority_key: Callable[[V], int] | None = None
) -> None

source

name

property name() -> str

source

Registry name for identification.

allow_overwrite

property allow_overwrite() -> bool

source

Whether overwriting existing keys is allowed.

freeze

def freeze() -> None

source

Prevent further registrations after application boot.

Once frozen, calls to register, register_factory, unregister, and clear will raise RegistryAlreadyExistsError. Read operations (get, resolve, has, etc.) are unaffected.

Call this from Provider.boot() or Application.boot() after all providers have completed their register() phase.

is_frozen

property is_frozen() -> bool

source

Return whether this registry has been frozen.

def register(
    key: K,
    value: V | None = None,
    *,
    allow_overwrite: bool | None = None
) -> V | Callable[[V], V] | None

source

register_factory

def register_factory(
    key: K,
    factory: Callable[[], V],
    *,
    allow_overwrite: bool | None = None
) -> None

source

unregister

def unregister(key: K) -> V | None

source

Unregister and return an item.

get

def get(key: K) -> V

source

get

def get(
    key: K,
    default: V
) -> V

source

get

def get(
    key: K,
    default: V | None = None
) -> V | None

source

Retrieve an item by key.

resolve

def resolve(key: K) -> V

source

Resolve an item, raising if not found.

get_or_raise

def get_or_raise(key: K) -> V

source

Alias for resolve.

has

def has(key: K) -> bool

source

Check if key is registered (item or factory).

keys

def keys() -> Iterable[K]

source

Return all registered keys (items only).

values

def values() -> Iterable[V]

source

Return all registered values.

values_ordered

def values_ordered() -> list[V]

source

Return all registered values sorted by priority.

If a priority_key was provided during initialization, values are sorted by that key in ascending order (lower values first). Otherwise, returns values in insertion order.

Returns

Type	Description
list[V]	List of values sorted by priority, or insertion order if no priority_key is configured.

items

def items() -> Iterable[tuple[K, V]]

source

Return all registered items.

factories

def factories() -> Iterable[K]

source

Return keys with registered factories.

all_keys

def all_keys() -> set[K]

source

Return all keys (items + factories).

clear

def clear() -> None

source

Clear all registered items and factories.

register_decorator

def register_decorator(
    key: K | None = None,
    *,
    allow_overwrite: bool | None = None
) -> Callable[[V], V]

source

Decorator for easy registration.

`RequestContext`

source

Context manager that sets request-scoped variables and restores them on exit.

__init__

def __init__(
    registry: ContextVarRegistry,
    *,
    request_id: str | None = None,
    method: str | None = None,
    path: str | None = None,
    correlation_id: str | None = None,
    causation_id: str | None = None,
    user_id: str | None = None,
    tenant_id: str | None = None,
    start_time: float | None = None,
    trace_id: str | None = None,
    span_id: str | None = None,
    trace_flags: str | None = None
) -> None

source

`Result`

source

Base Result type. Not abstract — Ok and Err are the only variants.

is_ok

def is_ok() -> bool

source

is_err

def is_err() -> bool

source

unwrap

def unwrap() -> T

source

unwrap_err

def unwrap_err() -> E

source

unwrap_or

def unwrap_or(default: T) -> T

source

unwrap_or_else

def unwrap_or_else(op: Callable[[E], T]) -> T

source

map_sync

def map_sync(op: Callable[[T], U]) -> Result[U, E]

source

map_err

def map_err(op: Callable[[E], F]) -> Result[T, F]

source

and_then_sync

def and_then_sync(op: Callable[[T], Result[U, E]]) -> Result[U, E]

source

or_else_sync

def or_else_sync(op: Callable[[E], Result[T, F]]) -> Result[T, F]

source

expect

def expect(message: str) -> T

source

match

def match(
    ok: Callable[[T], U],
    err: Callable[[E], U]
) -> U

source

map

async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

source

and_then

async def and_then(op: Callable[[T], Awaitable[Result[U, E]]]) -> Result[U, E]

source

or_else

async def or_else(op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]

source

flatten

def flatten() -> Result[Any, E]

source

filter

def filter(
    predicate: Callable[[T], bool],
    error: E
) -> Result[T, E]

source

ok_or

def ok_or(default: U) -> T | U

source

from_exception

def from_exception(
    cls,
    exc: Exception,
    ok_type: type[T] = type(None)
) -> Result[T, Exception]

source

Wrap a caught exception into an Err result.

to_optional

def to_optional() -> T | None

source

inspect

def inspect(op: Callable[[T], None]) -> Result[T, E]

source

inspect_err

def inspect_err(op: Callable[[E], None]) -> Result[T, E]

source

`ResultPipeline`

source

Pipeline for chaining Result operations with a fluent, type-safe API.

The error type E is preserved through the entire chain so mypy can verify error handling at every step. Use pipeline to start a pipeline from an infallible initial value.

__init__

def __init__(initial: Result[T, E]) -> None

source

then

def then(func: Callable[[T], Result[U, E]]) -> ResultPipeline[U, E]

source

Chain a function that returns a Result, preserving the error type.

map

def map(func: Callable[[T], U]) -> ResultPipeline[U, E]

source

Apply a function to the value if Ok, preserving the error type.

recover

def recover(func: Callable[[E], Result[T, F]]) -> ResultPipeline[T, F]

source

Recover from an error by applying a function that may produce a new error type.

finalize

def finalize() -> Result[T, E]

source

Get the final typed Result.

`Scope`

source

Request-scoped resolution context.

Caches scoped service instances for the lifetime of the scope. Singletons and transients delegate to the root container.

Supports nested scopes (K-02): parent scopes can be specified to create hierarchical scoping where child scopes can access parent scope services.

Usage

async with container.create_scope() as scope:
    session = await scope.resolve(DbSession)
    # same instance within this scope

async with container.create_scope() as scope:
    session = await scope.resolve(DbSession)
    # same instance within this scope

Nested scope usage

async with container.create_scope() as parent_scope:
    # parent scope
    async with parent_scope.create_scope() as child_scope:
        # child scope can access parent services
        session = await child_scope.resolve(DbSession)

async with container.create_scope() as parent_scope:
    # parent scope
    async with parent_scope.create_scope() as child_scope:
        # child scope can access parent services
        session = await child_scope.resolve(DbSession)

__init__

def __init__(
    root: Container,
    parent: Scope | None = None
) -> None

source

resolve

async def resolve(service_type: type[T]) -> T

source

Resolve a service from this scope.

K-02: Falls back to parent scope if not found in current scope.

create_scope

def create_scope() -> Scope

source

K-02: Create a child scope nested within this scope.

The child scope can access services from both the root container and the parent scope.

call

async def call(
    func: Callable[Ellipsis, Awaitable[T] | T],
    *args: Any,
    **kwargs: Any
) -> T

source

Call a function with dependency injection (Protocol implementation).

has

def has(service_type: object) -> bool

source

Check if a service is registered (delegates to root container).

resolve_optional

async def resolve_optional(service_type: type[T]) -> T | None

source

Resolve a service, returning None if not registered.

resolve_all

async def resolve_all(service_type: type[T]) -> list[T]

source

Resolve all registered implementations that are subtypes of a service type.

dispose

async def dispose() -> None

source

Dispose all scoped resources.

`SerializationError`

source

JSON serialisation or deserialisation failed.

Raised by:

JsonSerializer implementations
Domain model serialisation
Schema validation during decode

Functions

`SystemClock`

SystemClock

def SystemClock() -> ClockProtocol

source

Return the active clock (rarely needed by callers).

`as_result`

as_result

def as_result(*exceptions: type[Exception]) -> Callable[[Callable[P, Awaitable[T]]], Callable[P, Awaitable[Result[T, Exception]]]]

source

Decorator to capture exceptions and return a ``Result`` — **async functions** (canonical).

For sync functions use as_result_sync.

At least one exception type must be specified to prevent silently wrapping unexpected infrastructure failures.

Parameters

Parameter	Type	Description

Raises

Exception	Description
TypeError	If called with no exception types.

Example

@as_result(IOError, TimeoutError)
async def fetch(url: str) -> bytes:
    return await client.get(url)

@as_result(IOError, TimeoutError)
async def fetch(url: str) -> bytes:
    return await client.get(url)

`as_result_sync`

as_result_sync

def as_result_sync(*exceptions: type[Exception]) -> Callable[[Callable[P, T]], Callable[P, Result[T, Exception]]]

source

Decorator to capture exceptions and return a ``Result`` — **sync functions only**.

For async functions use as_result (the canonical async variant).

At least one exception type must be specified to prevent silently wrapping unexpected infrastructure failures.

Parameters

Parameter	Type	Description

Raises

Exception	Description
TypeError	If called with no exception types.

Example

@as_result_sync(ValueError, KeyError)
def parse(data: str) -> int:
    return int(data)

@as_result_sync(ValueError, KeyError)
def parse(data: str) -> int:
    return int(data)

`buildable`

buildable

def buildable(cls: type[T]) -> type[T]

source

Class decorator that auto-generates a fluent Builder for the decorated class.

Introspects __init__ to discover parameters and generates with_<param>() setter methods on a dynamically created builder. Required parameters (those without defaults) are validated at build() time and raise TypeError if missing.

Adds a .builder() class method that returns a fresh builder instance each time it is called.

Example

@buildable
class User:
    def __init__(self, name: str, email: str, age: int = 0):
        self.name = name
        self.email = email
        self.age = age

user = (
    User.builder()
    .with_name("John")
    .with_email("john@example.com")
    .with_age(30)
    .build()
)

@buildable
class User:
    def __init__(self, name: str, email: str, age: int = 0):
        self.name = name
        self.email = email
        self.age = age

user = (
    User.builder()
    .with_name("John")
    .with_email("john@example.com")
    .with_age(30)
    .build()
)

Parameters

Parameter	Type	Description
`cls`	type[T]	The class to decorate. Must have a standard ``__init__`` with type-annotated parameters.

Returns

Type	Description
type[T]	The same class with an attached ``.builder()`` static method.

`builder_field`

builder_field

def builder_field(field_or_func: str | Callable[Ellipsis, Any]) -> Callable[Ellipsis, Any]

source

Decorator or helper to define builder methods.

Can be used as a decorator on a method to mark it as a builder field, or as a metadata marker for automated builder generation.

`collect`

collect

def collect(results: Iterable[Result[T, E]]) -> Result[list[T], E]

source

Collect an iterable of Results into a Result of a list.

`create_app`

create_app

def create_app(
    name: str = 'lexigram-app',
    config: LexigramConfig | None = None
) -> Application

source

Create and return a new Lexigram application.

Returns a Application with no providers registered. Call Application.add_provider (or Application.add_providers) before Application.boot to register providers.

For entry-point-based auto-discovery, use lexigram.plugins.discovery.discover_providers from the lexigram-plugins package.

Parameters

Parameter	Type	Description
`name`	str	Human-readable application name used in log output. Defaults to ``"lexigram-app"``.
`config`	LexigramConfig \| None	Application configuration. If omitted an empty LexigramConfig is created, which reads values from environment variables via the standard ``LX_*`` prefix.

Returns

Type	Description
Application	A new Application ready for provider registration and boot.

Example

app = create_app(config=LexigramConfig.from_env_profile("production"))
app.add_provider(WebProvider(web_config))
await app.boot()

app = create_app(config=LexigramConfig.from_env_profile("production"))
app.add_provider(WebProvider(web_config))
await app.boot()

`create_module`

create_module

def create_module(
    name: str | None = None,
    *,
    providers: list[type] | None = None,
    imports: list[type | DynamicModule] | None = None,
    exports: list[type] | None = None,
    controllers: list[type] | None = None,
    is_global: bool = False,
    scan: list[str] | None = None,
    require_stub: bool = False
) -> Any

source

Create a module via a decorator — convenience alias for module.

`inject`

inject

def inject(func_or_cls: Any) -> Any

source

Decorator to enable automatic dependency injection.

Applied to a class: marks the class for constructor injection. Applied to an async function: wraps it to resolve missing arguments from the current DI container when invoked via container.call().

Parameters

Parameter	Type	Description
`func_or_cls`	Any	The class or async function to decorate.

Returns

Type	Description
Any	The decorated class or function with automatic DI support.

Raises

Exception	Description
TypeError	If applied to a synchronous (non-async) function.

`injectable`

injectable

def injectable(
    scope_or_cls: ServiceScope | type[T] = ServiceScope.TRANSIENT,
    name: str | None = None,
    *,
    scope: ServiceScope | None = None
) -> Callable[[type[T]], type[T]] | type[T]

source

Decorator to mark a class as injectable.

Parameters

Parameter	Type	Description
`scope_or_cls`	ServiceScope \| type[T]	The service scope or the class if used without parens.
`name`	str \| None	Optional name for the service registration.
`scope`	ServiceScope \| None	Explicit scope (keyword-only alternative to scope_or_cls).

Returns

Type	Description
Callable[[type[T]], type[T]] \| type[T]	The decorator function or the decorated class.

`module`

module

def module(
    cls: type[T],
    *,
    name: str | None = ...,
    providers: list[Any] | None = ...,
    imports: list[type | DynamicModule] | None = ...,
    exports: list[type] | None = ...,
    controllers: list[type] | None = ...,
    is_global: bool = ...,
    scan: list[str] | None = ...,
    require_stub: bool = ...
) -> type[T]

source

`partition`

partition

def partition(results: Iterable[Result[T, E]]) -> tuple[list[T], list[E]]

source

Partition results into successes and failures.

`provide`

provide

def provide(func: Callable[Ellipsis, Any]) -> type[FunctionProvider]

source

`request_context`

request_context

def request_context(
    registry: ContextVarRegistry,
    *,
    request_id: str | None = None,
    method: str | None = None,
    path: str | None = None,
    correlation_id: str | None = None,
    causation_id: str | None = None,
    user_id: str | None = None,
    tenant_id: str | None = None
) -> Generator[RequestContext, None, None]

source

Functional context manager for request-scoped context

with request_scope(registry, request_id=“abc”) as req_ctx: …

`run_application`

run_application

async def run_application(app: Application) -> None

source

Run an application with signal handling.

Starts the application, waits for SIGINT/SIGTERM, then performs a graceful shutdown. Handles :exc:KeyboardInterrupt and :exc:SystemExit silently so the process terminates cleanly.

Parameters

Parameter	Type	Description
`app`	Application	The Application instance to run.

`scoped`

scoped

def scoped(cls: type[T]) -> type[T]

source

Decorator to register a class as a scoped service.

A scoped service creates one instance per scope (e.g., per request).

Parameters

Parameter	Type	Description
`cls`	type[T]	The class to register as scoped.

Returns

Type	Description
type[T]	The decorated class marked for scoped registration.

`singleton`

singleton

def singleton(cls: type[T]) -> type[T]

source

Decorator to register a class as a singleton service.

A singleton service is created once and shared across all resolutions.

Parameters

Parameter	Type	Description
`cls`	type[T]	The class to register as a singleton.

Returns

Type	Description
type[T]	The decorated class marked for singleton registration.

`transient`

transient

def transient(cls: type[T]) -> type[T]

source

Decorator to register a class as a transient service.

A transient service creates a new instance each time it is resolved.

Parameters

Parameter	Type	Description
`cls`	type[T]	The class to register as transient.

Returns

Type	Description
type[T]	The decorated class marked for transient registration.

`try_catch`

try_catch

def try_catch(
    catch: tuple[type[Exception], Ellipsis],
    func: Callable[P, Awaitable[T]],
    *args: P.args,
    **kwargs: P.kwargs
) -> Awaitable[Result[T, Exception]]

source

Execute an async function and return its result as Ok, or Err if a listed exception occurs.

This is the canonical (async-first) variant. Callers must declare which exception types are expected failures; unexpected exceptions propagate normally.

For sync functions use try_catch_sync.

Parameters

Parameter	Type	Description
`catch`	tuple[type[Exception], Ellipsis]	A tuple of exception types to catch and wrap in ``Err``.
`func`	Callable[P, Awaitable[T]]	The async callable to invoke. args: Positional arguments forwarded to func. kwargs: Keyword arguments forwarded to func*.

Example

result = await try_catch((IOError, TimeoutError), fetch, url)

result = await try_catch((IOError, TimeoutError), fetch, url)

`try_catch_sync`

try_catch_sync

def try_catch_sync(
    catch: tuple[type[Exception], Ellipsis],
    func: Callable[P, T],
    *args: P.args,
    **kwargs: P.kwargs
) -> Result[T, Exception]

source

Execute a sync function and return its result as Ok, or Err if a listed exception occurs.

Unlike a bare try/except Exception, callers must explicitly declare which exception types are expected failures. Unexpected exceptions propagate normally.

For async functions use try_catch (the canonical async variant).

Parameters

Parameter	Type	Description
`catch`	tuple[type[Exception], Ellipsis]	A tuple of exception types to catch and wrap in ``Err``.
`func`	Callable[P, T]	The callable to invoke. args: Positional arguments forwarded to func. kwargs: Keyword arguments forwarded to func*.

Example

result = try_catch_sync((ValueError, KeyError), parse_int, "42")

result = try_catch_sync((ValueError, KeyError), parse_int, "42")

Exceptions

`ContainerError`

source

Base container error.

__init__

def __init__(
    message: str = 'Container error',
    **kwargs: Any
) -> None

source

`DomainError`

source

Business/domain-level errors (validation failures, not found, auth).

__init__

def __init__(
    message: str = 'Domain error',
    **kwargs: Any
) -> None

source

`InfrastructureError`

source

System-level failures (DB down, network timeouts).

__init__

def __init__(
    message: str = 'Infrastructure error',
    **kwargs: Any
) -> None

source

`LexigramError`

source

Root exception for the entire Lexigram ecosystem.

All Lexigram exceptions inherit from this class. This ensures that isinstance checks work identically across core, events, web, and all subpackages.

Attributes: code: Machine-readable error code (e.g., ‘LEX_CONTAINER_001’) message: Human-readable error message details: Additional context dictionary cause: Original exception that triggered this one hint: Optional suggestion for fixing the error

__init__

def __init__(
    message: str | None = None,
    details: dict[str, Any] | None = None,
    cause: BaseException | None = None,
    hint: str | None = None,
    **kwargs: Any
) -> None

source

docs_url

property docs_url() -> str

source

Return the documentation URL for this error code.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize exception to dictionary for logging/API responses.

with_details

def with_details(**kwargs: Any) -> Self

source

Return self with additional detail key-value pairs merged in.

Mutates and returns self to preserve the exception subtype.

with_hint

def with_hint(hint: str) -> Self

source

Return self with a human-readable hint attached.

with_cause

def with_cause(cause: BaseException) -> Self

source

Return self with __cause__ set to cause.

format

def format() -> str

source

Format for developer-friendly console output.

Returns a multi-line string containing the error class name, code, message, details, hint, and cause chain when present.

`LexigramException`

source

Base exception for all lexigram domain errors.

Extends LexigramError so that isinstance(exc, LexigramError) holds for every framework exception, enabling unified handling across all packages.

Use for catchable, domain-level failures such as configuration validation, dependency-injection resolution, input validation, and domain-model invariant violations. For infrastructure failures (database, network, I/O), raise the exception directly without wrapping it in this hierarchy.

Example

try:
    service = await container.resolve(MyService)
except LexigramException as e:
    logger.error("resolution_failed", error=str(e))
    raise

try:
    service = await container.resolve(MyService)
except LexigramException as e:
    logger.error("resolution_failed", error=str(e))
    raise

`ModuleError`

source

Module configuration or lifecycle error.

__init__

def __init__(
    message: str = 'Module error',
    **kwargs: Any
) -> None

source

`NotFoundError`

source

Resource not found error (domain-level).

__init__

def __init__(
    message: str = 'Not found',
    **kwargs: Any
) -> None

source

`ProviderError`

source

Provider configuration or lifecycle error.

__init__

def __init__(
    message: str = 'Provider error',
    **kwargs: Any
) -> None

source