API Reference
Protocols
Section titled “Protocols”CacheBackendProtocol
Section titled “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)))async def get(key: str) -> Result[Any | None, CacheError]
Get a value from the cache.
| Parameter | Type | Description |
|---|---|---|
key | str | The cache key to retrieve. |
| 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.
async def set( key: str, value: Any, ttl: int | None = None ) -> Result[None, CacheError]
Set a value in the cache with optional TTL.
| 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). |
| Type | Description |
|---|---|
| Result[None, CacheError] | Ok(None) if successful, Err(CacheError) on failure. |
async def delete(key: str) -> Result[bool, CacheError]
Delete a value from the cache.
| Parameter | Type | Description |
|---|---|---|
key | str | The cache key to delete. |
| Type | Description |
|---|---|
| Result[bool, CacheError] | Ok(True) if deleted, Ok(False) if not found, Err(CacheError) on failure. |
async def delete_many(keys: list[str]) -> Result[int, CacheError]
Delete multiple values from the cache.
| Parameter | Type | Description |
|---|---|---|
keys | list[str] | List of cache keys to delete. |
| Type | Description |
|---|---|
| Result[int, CacheError] | Ok(count) of deleted keys, Err(CacheError) on failure. |
async def delete_pattern(pattern: str) -> Result[int, CacheError]
Delete all keys matching a glob-style pattern.
| Parameter | Type | Description |
|---|---|---|
pattern | str | Glob pattern (e.g. `"pet:list:*"). Supports *` as a wildcard matching any sequence of characters. |
| Type | Description |
|---|---|
| Result[int, CacheError] | Ok(count) of deleted keys, Err(CacheError) on failure. |
async def exists(key: str) -> Result[bool, CacheError]
Check if a key exists in the cache.
| Parameter | Type | Description |
|---|---|---|
key | str | The cache key to check. |
| Type | Description |
|---|---|
| Result[bool, CacheError] | Ok(True) if exists, Ok(False) otherwise, Err(CacheError) on failure. |
async def clear() -> Result[None, CacheError]
Clear all values from the cache.
| Type | Description |
|---|---|
| Result[None, CacheError] | Ok(None) if successful, Err(CacheError) on failure. |
async def get_many(keys: list[str]) -> Result[dict[str, Any], CacheError]
Get multiple values from the cache.
| Parameter | Type | Description |
|---|---|---|
keys | list[str] | List of cache keys to retrieve. |
| Type | Description |
|---|---|
| Result[dict[str, Any], CacheError] | Ok(dict) mapping found keys to values, Err(CacheError) on failure. |
async def set_many( items: dict[str, Any], ttl: int | None = None ) -> Result[None, CacheError]
Set multiple values in the cache.
| 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. |
| Type | Description |
|---|---|
| Result[None, CacheError] | Ok(None) if all items set successfully, Err(CacheError) on failure. |
async def health_check(timeout: float = 5.0) -> HealthCheckResult
Perform a health check on the cache backend.
| Type | Description |
|---|---|
| HealthCheckResult | Structured HealthCheckResult. |
ClockProtocol
Section titled “ClockProtocol”Injectable time source for the framework.
def now() -> datetime
Return the current timezone-aware UTC time.
def monotonic() -> float
Return a monotonic elapsed-time counter.
def timestamp() -> float
Return the current Unix timestamp in UTC seconds.
def time() -> float
Backward-compatible alias for timestamp.
ConfigProtocol
Section titled “ConfigProtocol”Protocol for configuration access across the framework.
Any configuration object (BaseConfig, QuadkitConfig, 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)property environment() -> Environment
The active deployment environment.
property is_production() -> bool
Return True when the active environment is production.
property is_development() -> bool
Return True when the active environment is development.
property is_testing() -> bool
Return True when the active environment is testing.
property is_staging() -> bool
Return True when the active environment is staging.
property is_debug() -> bool
Return True when debug mode is enabled.
def get( key: str, default: Any = None ) -> Any
Get a configuration value by dot-notation key.
| Parameter | Type | Description |
|---|---|---|
key | str | Configuration key (e.g. `"app.name" or "database.url"`). |
default | Any | Value returned when the key is not found. |
| Type | Description |
|---|---|
| Any | The configuration value, or *default* if not found. |
def get_section( name: str, model_cls: None = None ) -> dict[str, Any] | None
def get_section( name: str, model_cls: type[T] ) -> T
def get_section( name: str, model_cls: type[T] | None = None ) -> T | dict[str, Any] | None
Get a typed configuration section.
| Parameter | Type | Description |
|---|---|---|
name | str | Section name (e.g. `"logging"`). |
model_cls | type[T] | None | Optional model class to coerce the section into. |
| Type | Description |
|---|---|
| T | dict[str, Any] | None | A model instance when *model_cls* is provided, otherwise a raw dict or the attribute value; `None` when the section is absent and no model class was supplied. |
def has_section(name: str) -> bool
Check whether a configuration section exists.
| Parameter | Type | Description |
|---|---|---|
name | str | Section name to check. |
| Type | Description |
|---|---|
| bool | True if the section is present. |
EventMiddlewareProtocol
Section titled “EventMiddlewareProtocol”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
Section titled “HookRegistryProtocol”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.
def register_action( hook_name: str, handler: Callable[Ellipsis, Any], priority: int = 100, *, once: bool = False ) -> None
Register an action handler for the given hook.
| 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. |
def register_filter( hook_name: str, handler: Callable[Ellipsis, Any], priority: int = 100, *, once: bool = False ) -> None
Register a filter handler for the given hook.
| 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. |
def unregister_action( hook_name: str, handler: Callable[Ellipsis, Any] ) -> bool
Remove an action handler. Returns True if found and removed.
def unregister_filter( hook_name: str, handler: Callable[Ellipsis, Any] ) -> bool
Remove a filter handler. Returns True if found and removed.
async def call_action( hook_name: str, **kwargs: Any ) -> None
Invoke all action handlers. Errors are isolated and logged.
async def apply_filter( hook_name: str, value: Any, **kwargs: Any ) -> Any
Pass value through all filter handlers. Errors propagate.
def has_action(hook_name: str) -> bool
Check if any action handlers are registered.
def has_filter(hook_name: str) -> bool
Check if any filter handlers are registered.
def clear(hook_name: str | None = None) -> None
Clear handlers for a specific hook, or all hooks.
IdGeneratorProtocol
Section titled “IdGeneratorProtocol”Injectable identity generator.
property strategy() -> IdStrategy
Return the active ID generation strategy.
def generate() -> str
Generate a new unique identifier.
def generate_for(entity_type: str) -> str
Generate a new identifier for a specific entity type.
LoggerFactoryProtocol
Section titled “LoggerFactoryProtocol”Protocol for creating/retrieving named logger instances.
def get_logger(name: str | None = None) -> LoggerProtocol
Get a logger instance for the given name.
LoggerProtocol
Section titled “LoggerProtocol”Protocol for structured logging with context support.
The message is positional-only. Every keyword reaching a log call is a
structured field, so binding one of them to the message parameter
would reserve an ordinary word (msg) that callers are entitled to
log — and logger.info("mail_sent", msg=body) would fail with
“got multiple values for argument ‘msg’”.
def debug( *args: Any, **kwargs: Any ) -> None
def info( *args: Any, **kwargs: Any ) -> None
def warning( *args: Any, **kwargs: Any ) -> None
def error( *args: Any, **kwargs: Any ) -> None
def critical( *args: Any, **kwargs: Any ) -> None
def exception( *args: Any, **kwargs: Any ) -> None
def bind(**kwargs: Any) -> LoggerProtocol
Add context variables to the logger.
def unbind(*keys: str) -> LoggerProtocol
Remove context variables from the logger.
RedactorProtocol
Section titled “RedactorProtocol”Protocol for sensitive data redaction.
def redact_dict(data: dict[str, Any]) -> dict[str, Any]
Return a copy of data with sensitive values masked.
def redact_value(value: Any) -> Any
Redact a single value if it matches a sensitive pattern.
Classes
Section titled “Classes”AbstractBuilder
Section titled “AbstractBuilder”Base class for all fluent builders in Quadkit.
Provides foundational support for state management and fluent chaining.
def __init__() -> None
Initialize builder state.
def copy() -> Self
Return a deep copy of the builder state.
Useful for creating slightly different variations of an object without rebuilding from scratch.
def build() -> T
Construct the final object from the builder state.
| Type | Description |
|---|---|
| T | The constructed object of type T. |
AppState
Section titled “AppState”Application lifecycle states.
Application
Section titled “Application”The composition root.
Usage
async with Application.boot(providers=[MyProvider()]) as app: # resolve Invoker from the container for entry-point injection from quadkit.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 quadkit.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 quadkit.app import run_applicationawait run_application(app)app = Application()app.add_provider(MyProvider())from quadkit.app import run_applicationawait run_application(app)def __init__( name: str = 'quadkit-app', config: QuadkitConfig | None = None ) -> None
property state() -> AppState
Current application state.
property is_running() -> bool
Check if application is running.
property logger() -> LoggerProtocol
Get the application logger.
property config() -> QuadkitConfig
Application configuration.
property container() -> Container
DI container.
property module_graph() -> CompiledModuleGraph | None
The compiled module graph, or None before compilation.
Read-only view of what the boot sequence compiled: every module with
its providers, imports, and fully expanded exports. It exists so
boot-time guards can assert that what a module declares it exports is
what the container can actually resolve — the failure this graph is
otherwise blind to, because an export that nothing resolves during
boot only fails later, at QK_ERR_DI_004 on first use, or never.
| Type | Description |
|---|---|
| CompiledModuleGraph | None | The compiled graph, or `None` when the application has not been compiled yet. |
property providers() -> list[Provider]
Get all registered providers.
def add_middleware(middleware: MiddlewareProtocol[Any, Any]) -> None
Add a middleware to the application.
def add_provider(provider: Provider) -> None
Add a provider to be orchestrated.
Must be called before start().
def add_providers(providers: list[Provider]) -> None
Add multiple providers. Must be called before start().
def discover_providers(*packages: str) -> None
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.
| Parameter | Type | Description |
|---|
Example
app.discover_providers("my_platform.modules", "my_platform.infrastructure")app.discover_providers("my_platform.modules", "my_platform.infrastructure")def add_module(module: type | Any) -> None
Add a module to the application.
Modules organize providers and define visibility boundaries.
Must be called before start().
def add_modules(modules: Iterable[type | Any]) -> None
Register modules in iterable order without mutating the collection.
Each item retains the same pre-start guard as add_module.
def discover_modules( *, entry_point_group: str = 'quadkit.modules', directories: list[str | Path] | None = None, enabled: list[str] | None = None, disabled: list[str] | None = None ) -> None
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().
| Parameter | Type | Description |
|---|---|---|
entry_point_group | str | Entry-point group to scan. Default "quadkit.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"])async def start() -> None
Boot all providers and start the application.
Phases:
0. Validate configuration against the active environment
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.
| Exception | Description |
|---|---|
| RuntimeError | If application is not in CREATED state. |
| ConfigurationError | If the configuration violates environment-specific constraints (e.g. `debug=True` in production). |
async def stop() -> None
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.
async def startup_check(timeout: float | None = None) -> AggregateHealthResult
Run startup checks for the application.
async def boot( cls, name: str = 'quadkit-app', providers: list[Provider] | None = None, modules: list[Any] | None = None, config: QuadkitConfig | None = None ) -> AsyncIterator[Application]
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
Section titled “ConfigProvider”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.
def __init__( config_class: type[Any] = QuadkitConfig, sources: list[Any] | None = None, env_prefix: str = '', extensions: dict[str, type] | None = None ) -> None
async def register(container: ContainerRegistrarProtocol) -> None
Load configuration and register as a singleton.
async def boot(container: ContainerResolverProtocol) -> None
No boot-time work required for the config module.
async def shutdown() -> None
No resources to release for the config module.
async def health_check(timeout: float = 5.0) -> HealthCheckResult
Report config loading status.
ConfigurationError
Section titled “ConfigurationError”Configuration loading, parsing, or validation failed.
Raised by:
ConfigProviderduring initialisation- Config loaders when sources are invalid
- Validation of
ConfigProtocolimplementations
Container
Section titled “Container”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)def __init__( parent: Container | None = None, testing_mode: bool = False, strict_mode: bool = False, validate_on_register: bool = True ) -> None
property registrar() -> ContainerRegistrarImpl
Return the write-only registrar component.
property resolver() -> ContainerResolverImpl
Return the read-only resolver component.
def clear() -> None
Clear all registered services.
| Exception | Description |
|---|---|
| ContainerError | if the container has been frozen. |
def transient( service_type: type[T], factory: ServiceFactory[T], validate: bool | None = None ) -> None
Register a transient service (new instance each resolution).
validate=None (default) falls back to
validate_on_register.
def singleton( service_type: type[T], instance: T | None = None, *, name: str | None = None, factory: ServiceFactory[T] | None = None, validate: bool = True ) -> None
def singleton( service_type: Any, instance: Any = None, *, name: str | None = None, factory: Any | None = None, validate: bool = True ) -> None
def singleton( service_type: Any, instance: Any = None, *, name: str | None = None, factory: Any | None = None, validate: bool = True ) -> None
Register a singleton service (shared instance).
Pass instance for a pre-built singleton, or factory for lazy creation.
Use name for named registrations resolvable via Annotated[T, Named('...')]].
def scoped( service_type: type[T], factory: ServiceFactory[T], validate: bool | None = None, *, name: str | None = None ) -> None
Register a scoped service (instance per scope).
validate=None (default) falls back to
validate_on_register.
property is_frozen() -> bool
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.
def freeze(validate: bool = True) -> None
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()).
| 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. |
| Exception | Description |
|---|---|
| ContainerValidationError | If `validate=True` and validation issues are found. |
ContainerBuilder
Section titled “ContainerBuilder”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())def __init__() -> None
Initialize the container builder.
def add_singleton( interface: type[T], implementation: type[T] | Callable[[], T], metadata: dict[str, Any] | None = None ) -> Self
Register a singleton service.
| 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. |
| Type | Description |
|---|---|
| Self | Self for method chaining. |
def add_instance( interface: type[T], instance: T ) -> Self
Register a pre-built instance as a singleton.
| Parameter | Type | Description |
|---|---|---|
interface | type[T] | The service interface/contract. |
instance | T | The pre-built instance. |
| Type | Description |
|---|---|
| Self | Self for method chaining. |
def add_scoped( interface: type[T], implementation: type[T] | Callable[Ellipsis, T], metadata: dict[str, Any] | None = None ) -> Self
Register a scoped service.
| 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. |
| Type | Description |
|---|---|
| Self | Self for method chaining. |
def add_transient( interface: type[T], implementation: type[T] | Callable[Ellipsis, T], metadata: dict[str, Any] | None = None ) -> Self
Register a transient service.
| 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. |
| Type | Description |
|---|---|
| Self | Self for method chaining. |
def add_module(module_or_dynamic: type | DynamicModule) -> Self
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.
| Parameter | Type | Description |
|---|---|---|
module_or_dynamic | type | DynamicModule | A module class or DynamicModule. |
| Type | Description |
|---|---|
| Self | Self for method chaining. |
def add_provider(provider: Any) -> Self
Add a provider whose register() will be called during build.
| Parameter | Type | Description |
|---|---|---|
provider | Any | A Provider instance. |
| Type | Description |
|---|---|
| Self | Self for method chaining. |
def disable_validation() -> Self
Disable container validation during build.
| Type | Description |
|---|---|
| Self | Self for method chaining. |
async def build() -> ContainerResolverProtocol
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)
| Type | Description |
|---|---|
| ContainerResolverProtocol | A ContainerResolverProtocol for resolving services. |
| Exception | Description |
|---|---|
| ContainerBuildError | If build or validation fails. |
Context
Section titled “Context”User-facing typed wrapper around an injected ``ContextVarRegistry``.
def __init__(registry: ContextVarRegistry) -> None
property registry() -> ContextVarRegistry
The underlying registry (for advanced use / wiring).
def get( key: ContextKey[T], default: T | None = None ) -> T | None
Get the current value for a typed key.
def set( key: ContextKey[T], value: T ) -> contextvars.Token[T | None]
Set a typed value, returning a reset token.
def reset( key: ContextKey[T], token: contextvars.Token[T | None] ) -> None
Reset a typed value using a token from set.
def register_key(key: ContextKey[Any]) -> None
Register a new context key at runtime.
def get_dynamic( key: str, default: Any = None ) -> Any
Get a value by string key.
def set_dynamic( key: str, value: Any ) -> contextvars.Token[Any]
Set a value by string key.
def reset_dynamic( key: str, token: contextvars.Token[Any] ) -> None
Reset a value by string key.
def snapshot() -> dict[str, Any]
Snapshot of all non-None context values.
def has(key: str) -> bool
Check whether a context key is registered.
ContextKey
Section titled “ContextKey”Immutable, typed key for context-variable access (pure data).
CoreInfrastructureProvider
Section titled “CoreInfrastructureProvider”Provider for **shared low-level Quadkit 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 propagationRegistry— 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 (QuadkitApp) |
+-----------------------------------+----------------------------------------------------+
| name = "common" | name = "core" |
+-----------------------------------+----------------------------------------------------+
async def register(container: ContainerRegistrarProtocol) -> None
Register common utilities with the container.
async def boot(container: ContainerResolverProtocol) -> None
Initialize common services.
async def shutdown() -> None
Clean up common resources.
async def health_check(timeout: float = 5.0) -> HealthCheckResult
Check health of shared infrastructure.
CoreModule
Section titled “CoreModule”Kernel provider manifest for a minimal Quadkit application.
Bundles the four kernel framework providers that every Quadkit 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 quadkit core package and contracts from
quadkit-contracts.
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]
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.
| Parameter | Type | Description |
|---|---|---|
config_class | type[Any] | None | Custom configuration dataclass to use instead of the default QuadkitConfig. |
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. |
| 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()},)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
Create a DynamicModule configured with core provider instances.
| 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. |
| Type | Description |
|---|---|
| DynamicModule | DynamicModule configured with instantiated core providers and default core exports. |
CoreProvider
Section titled “CoreProvider”Main **aggregating** provider for Quadkit 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" |
+-----------------------------------+----------------------------------------------------+
def __init__() -> None
Initialize CoreProvider.
async def register(container: ContainerRegistrarProtocol) -> None
Register core services.
Note: Sub-providers are now handled by the orchestrator via the sub_providers property and recursive discovery.
async def boot(container: ContainerResolverProtocol) -> None
Boot core services.
async def shutdown() -> None
Shutdown core services.
async def health_check(timeout: float = 5.0) -> HealthCheckResult
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
Section titled “DomainModelError”Domain model invariant violated or business rule broken.
Raised by:
- Domain event handlers
- Aggregate root validations
- Entity lifecycle violations
DynamicModule
Section titled “DynamicModule”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.
def __init__(error: E) -> None
def is_ok() -> bool
def is_err() -> bool
def unwrap() -> T
def unwrap_err() -> E
def unwrap_or(default: T) -> T
def unwrap_or_else(op: Callable[[E], T]) -> T
def map_sync(op: Callable[[T], U]) -> Result[U, E]
def map_err(op: Callable[[E], F]) -> Result[T, F]
def expect(message: str) -> T
def match( ok: Callable[[T], U], err: Callable[[E], U] ) -> U
async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]
async def async_map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]
Alias for map — exists for backward compatibility.
def flatten() -> Result[Any, E]
def filter( predicate: Callable[[T], bool], error: E ) -> Result[T, E]
def ok_or(default: U) -> U
EventStream
Section titled “EventStream”An async-iterable stream of items, composable with ``pipe``.
def pipe(*operators: Op[Any, Any]) -> EventStream[Any]
Apply operators left to right, returning a new stream.
HookPriority
Section titled “HookPriority”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
Section titled “HookRegistry”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.
| 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")def __init__(name: str = 'default') -> None
def register_action( hook_name: str, handler: Callable[Ellipsis, Any], priority: int = HookPriority.NORMAL, *, once: bool = False ) -> None
Register an action handler for the given hook.
| 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. |
def register_filter( hook_name: str, handler: Callable[Ellipsis, Any], priority: int = HookPriority.NORMAL, *, once: bool = False ) -> None
Register a filter handler for the given hook.
Filter handlers receive a value as the first positional argument and must return the (possibly transformed) value.
| 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. |
def unregister_action( hook_name: str, handler: Callable[Ellipsis, Any] ) -> bool
Remove an action handler from a hook.
| Parameter | Type | Description |
|---|---|---|
hook_name | str | The hook point name. |
handler | Callable[Ellipsis, Any] | The handler to remove (identity comparison). |
| Type | Description |
|---|---|
| bool | True if the handler was found and removed. |
def unregister_filter( hook_name: str, handler: Callable[Ellipsis, Any] ) -> bool
Remove a filter handler from a hook.
| Parameter | Type | Description |
|---|---|---|
hook_name | str | The hook point name. |
handler | Callable[Ellipsis, Any] | The handler to remove (identity comparison). |
| Type | Description |
|---|---|
| bool | True if the handler was found and removed. |
async def call_action( hook_name: str, **kwargs: Any ) -> None
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).
| Parameter | Type | Description |
|---|---|---|
hook_name | str | The hook point name. **kwargs: Arguments passed to each handler. |
async def apply_filter( hook_name: str, value: Any, **kwargs: Any ) -> Any
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.
| Parameter | Type | Description |
|---|---|---|
hook_name | str | The hook point name. |
value | Any | The initial value to filter. **kwargs: Additional arguments passed to each handler. |
| Type | Description |
|---|---|
| Any | The final filtered value. |
def has_action(hook_name: str) -> bool
Check if any action handlers are registered for a hook.
| Parameter | Type | Description |
|---|---|---|
hook_name | str | The hook point name. |
| Type | Description |
|---|---|
| bool | True if at least one handler is registered. |
def has_filter(hook_name: str) -> bool
Check if any filter handlers are registered for a hook.
| Parameter | Type | Description |
|---|---|---|
hook_name | str | The hook point name. |
| Type | Description |
|---|---|
| bool | True if at least one handler is registered. |
def action_names() -> list[str]
List all registered action hook names.
| Type | Description |
|---|---|
| list[str] | List of hook names that have at least one action handler. |
def filter_names() -> list[str]
List all registered filter hook names.
| Type | Description |
|---|---|
| list[str] | List of hook names that have at least one filter handler. |
def clear(hook_name: str | None = None) -> None
Clear all handlers, or handlers for a specific hook.
| Parameter | Type | Description |
|---|---|---|
hook_name | str | None | If provided, clear only this hook. Otherwise clear all. |
def action( hook_name: str, *, priority: int = HookPriority.NORMAL, once: bool = False ) -> Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]
Decorator to register an action handler.
Returns the handler unchanged so it can still be called directly.
| 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. |
| 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"])def filter( hook_name: str, *, priority: int = HookPriority.NORMAL, once: bool = False ) -> Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]
Decorator to register a filter handler.
Returns the handler unchanged so it can still be called directly.
| 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. |
| 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 valueIdentityProvider
Section titled “IdentityProvider”Register identity generators for the framework.
def __init__(config: IdentityConfig | None = None) -> None
async def register(container: ContainerRegistrarProtocol) -> None
Register the active generator implementation.
async def boot(container: BootContainerProtocol) -> None
Install the resolved generator into ambient.
async def shutdown() -> None
Identity services do not require shutdown work.
async def health_check(timeout: float = 5.0) -> HealthCheckResult
Report the identity subsystem as healthy.
Injectable
Section titled “Injectable”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.
| 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. |
def __init__( scope: ServiceScope = ServiceScope.TRANSIENT, name: str | None = None ) -> None
Initialize the Injectable decorator.
| Parameter | Type | Description |
|---|---|---|
scope | ServiceScope | The service scope for this class. |
name | str | None | Optional name for the service. |
InjectionError
Section titled “InjectionError”Dependency injection or container resolution failed.
Raised by:
Container.resolve()when a service is not registeredProvider.register()on configuration errors- Module resolution when imports fail
- Circular dependency detection
Invoker
Section titled “Invoker”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.
def __init__( container: Container, middleware: MiddlewarePipelineProtocol ) -> None
async def invoke( func: Callable[Ellipsis, Awaitable[T]], *args: Any, **kwargs: Any ) -> T
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
Section titled “LoggingProvider”Structured logging factory and helpers DI provider.
async def register(container: ContainerRegistrarProtocol) -> None
Register logging services.
async def boot(container: ContainerResolverProtocol) -> None
Apply logging configuration from QuadkitConfig.
async def shutdown() -> None
No resources to release for the logging module.
async def health_check(timeout: float = 5.0) -> HealthCheckResult
Health check — always healthy (in-process only, no external backend).
| Parameter | Type | Description |
|---|---|---|
timeout | float | Ignored for in-process providers. |
| Type | Description |
|---|---|
| HealthCheckResult | Always HEALTHY — no external backend to check. |
MiddlewareChain
Section titled “MiddlewareChain”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.
def __init__(middleware: list[Any] | None = None) -> None
Create a chain, optionally pre-populated with middleware.
| Parameter | Type | Description |
|---|---|---|
middleware | list[Any] | None | Initial list of middleware to seed the chain with. |
def add(middleware: Any) -> MiddlewareChain
Append a middleware and return self for chaining.
| Parameter | Type | Description |
|---|---|---|
middleware | Any | An `async (context, next_handler) -> result` callable. |
| Type | Description |
|---|---|
| MiddlewareChain | `self` so calls can be chained fluently. |
def build() -> MiddlewarePipeline
Produce an immutable MiddlewarePipeline from the current state.
The returned pipeline is a snapshot — subsequent add calls do not affect already-built pipelines.
| Type | Description |
|---|---|
| MiddlewarePipeline | A new MiddlewarePipeline containing all added middleware. |
def clear() -> None
Remove all middleware from the chain without building.
MiddlewareConfig
Section titled “MiddlewareConfig”Middleware chain, pipeline, and registry configuration.
MiddlewareModule
Section titled “MiddlewareModule”Composable async middleware pipeline, exception filter chain, and request pipeline.
Call configure to configure the middleware subsystem with custom settings.
Usage
from quadkit.middleware.config import MiddlewareConfig
@module( imports=[MiddlewareModule.configure(MiddlewareConfig(...))])class AppModule(Module): passfrom quadkit.middleware.config import MiddlewareConfig
@module( imports=[MiddlewareModule.configure(MiddlewareConfig(...))])class AppModule(Module): passdef configure( cls, config: Any | None = None ) -> DynamicModule
Create a MiddlewareModule with explicit configuration.
| Parameter | Type | Description |
|---|---|---|
config | Any | None | MiddlewareConfig or `None` for framework defaults. |
| Type | Description |
|---|---|
| DynamicModule | A DynamicModule descriptor. |
def stub( cls, config: Any | None = None ) -> DynamicModule
Return a MiddlewareModule with framework defaults for testing.
All middleware uses default configuration. No request pipeline is active — only the DI bindings are registered.
| Parameter | Type | Description |
|---|---|---|
config | Any | None | Accepted for interface compatibility with stub; ignored. Pass MiddlewareConfig to configure instead. |
| Type | Description |
|---|---|
| DynamicModule | A DynamicModule with default middleware configuration. |
MiddlewareProvider
Section titled “MiddlewareProvider”Middleware chain, pipeline, and registry DI provider.
On register():
- Registers
MiddlewareChain,ExceptionFilterChain, andMiddlewarePipelineas container singletons. - Scans the
quadkit.middlewareentry-point group and delegates registration to every discoveredProvidersubclass so that third-party middleware packages integrate automatically.
def __init__() -> None
async def register(container: ContainerRegistrarProtocol) -> None
Register middleware singletons and discover third-party providers.
| Parameter | Type | Description |
|---|---|---|
container | ContainerRegistrarProtocol | The DI container registrar. |
async def boot(container: BootContainerProtocol) -> None
Boot discovered providers after middleware chains are initialized.
| Parameter | Type | Description |
|---|---|---|
container | BootContainerProtocol | The DI container for boot phase. |
async def shutdown() -> None
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).
async def health_check(timeout: float = 5.0) -> HealthCheckResult
Check provider health.
| Parameter | Type | Description |
|---|---|---|
timeout | float | Maximum seconds to wait for health check response. |
| Type | Description |
|---|---|
| HealthCheckResult | HealthCheckResult with status and component details. |
MiddlewareRegistry
Section titled “MiddlewareRegistry”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()def __init__() -> None
Initialize the middleware registry with priority ordering.
def register( name: str, middleware: Any, *, priority: int = 50 ) -> None
Register a named middleware.
| 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. |
| Exception | Description |
|---|---|
| ValueError | If a middleware with the same name is already registered. |
def unregister(name: str) -> Any
Remove and return a named middleware.
| Parameter | Type | Description |
|---|---|---|
name | str | The middleware name to remove. |
| Type | Description |
|---|---|
| Any | The removed middleware instance. |
| Exception | Description |
|---|---|
| KeyError | If no middleware with the given name is registered. |
def get(name: str) -> Any
Retrieve a named middleware.
| Parameter | Type | Description |
|---|---|---|
name | str | The middleware name to look up. |
| Type | Description |
|---|---|
| Any | The middleware instance. |
| Exception | Description |
|---|---|
| KeyError | If no middleware with the given name is registered. |
def all() -> list[Any]
Return all middleware in priority order.
| Type | Description |
|---|---|
| list[Any] | List of middleware instances sorted by ascending priority. |
def has(name: str) -> bool
Check if a middleware with the given name is registered.
def names() -> list[str]
Return a sorted list of registered middleware names.
def build_chain() -> MiddlewareChain
Build a MiddlewareChain from all registered middleware.
Returns middleware in ascending priority order (lower priority number executes first).
| 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)def with_defaults(cls) -> MiddlewareRegistry
Pre-populate with standard infrastructure middleware.
def with_resilience( cls, config: MiddlewareConfig | None = None, catch: type[Exception] | tuple[type[Exception], Ellipsis] = (ConnectionError, TimeoutError) ) -> MiddlewareRegistry
Defaults plus resilience middlewares built from config.
Every MiddlewareConfig resilience field has a consumer here:
default_retry_count/default_retry_delay -> RetryMiddleware,
circuit_failure_threshold/circuit_recovery_timeout ->
CircuitBreakerMiddleware, rate_limit_max_requests/
rate_limit_window -> RateLimiterMiddleware,
default_timeout -> TimeoutMiddleware.
| Parameter | Type | Description |
|---|---|---|
config | MiddlewareConfig | None | Middleware defaults; falls back to `MiddlewareConfig()`. |
catch | type[Exception] | tuple[type[Exception], Ellipsis] | Exceptions RetryMiddleware retries (connection-level by default; widen deliberately, e.g. to `Exception`). |
Module
Section titled “Module”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 Quadkit 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): passClassVar 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)], )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
Optionally override ClassVar defaults with instance values.
| 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. |
def configure( cls, *args: Any, **kwargs: Any ) -> DynamicModule
Configure this module globally. Called once at the app root.
| Parameter | Type | Description |
|---|
| Type | Description |
|---|---|
| DynamicModule | DynamicModule for root import. |
| Exception | Description |
|---|---|
| NotImplementedError | Subclasses must override this method. |
def scope( cls, *providers: type, **kwargs: Any ) -> DynamicModule
Register additional providers in a per-feature scope.
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).
| Parameter | Type | Description |
|---|
| Type | Description |
|---|---|
| DynamicModule | DynamicModule with a unique anonymous token as module identity. |
def stub( cls, config: Any = None ) -> DynamicModule
Return a test-mode module with in-memory/noop backends.
| Parameter | Type | Description |
|---|---|---|
config | Any | Optional test configuration override. |
| Type | Description |
|---|---|
| DynamicModule | DynamicModule using test providers. |
| Exception | Description |
|---|---|
| NotImplementedError | Subclasses must override this method. |
async def on_module_booted(cls) -> None
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.
async def on_module_shutdown(cls) -> None
Called before this module’s providers are shut down.
Override in subclasses to run pre-shutdown cleanup logic.
ModuleBase
Section titled “ModuleBase”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 Quadkit 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): passClassVar 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)], )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
Optionally override ClassVar defaults with instance values.
| 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. |
def configure( cls, *args: Any, **kwargs: Any ) -> DynamicModule
Configure this module globally. Called once at the app root.
| Parameter | Type | Description |
|---|
| Type | Description |
|---|---|
| DynamicModule | DynamicModule for root import. |
| Exception | Description |
|---|---|
| NotImplementedError | Subclasses must override this method. |
def scope( cls, *providers: type, **kwargs: Any ) -> DynamicModule
Register additional providers in a per-feature scope.
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).
| Parameter | Type | Description |
|---|
| Type | Description |
|---|---|
| DynamicModule | DynamicModule with a unique anonymous token as module identity. |
def stub( cls, config: Any = None ) -> DynamicModule
Return a test-mode module with in-memory/noop backends.
| Parameter | Type | Description |
|---|---|---|
config | Any | Optional test configuration override. |
| Type | Description |
|---|---|
| DynamicModule | DynamicModule using test providers. |
| Exception | Description |
|---|---|
| NotImplementedError | Subclasses must override this method. |
async def on_module_booted(cls) -> None
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.
async def on_module_shutdown(cls) -> None
Called before this module’s providers are shut down.
Override in subclasses to run pre-shutdown cleanup logic.
def __init__(value: T) -> None
def is_ok() -> bool
def is_err() -> bool
def unwrap() -> T
def unwrap_err() -> E
def unwrap_or(default: T) -> T
def unwrap_or_else(op: Callable[[E], T]) -> T
def map_sync(op: Callable[[T], U]) -> Result[U, E]
def map_err(op: Callable[[E], F]) -> Result[T, F]
def expect(message: str) -> T
def match( ok: Callable[[T], U], err: Callable[[E], U] ) -> U
async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]
async def async_map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]
Alias for map — exists for backward compatibility.
def flatten() -> Result[Any, E]
def filter( predicate: Callable[[T], bool], error: E ) -> Result[T, E]
def ok_or(default: U) -> T
PluginEngineProvider
Section titled “PluginEngineProvider”Register every enabled entry-point provider declared by a plugin.
During register() the provider loads the plugin-state disabled
set and delegates discovery, filtering, and instantiation to the shared
discover_providers() primitive — the single implementation of the
entry-point discover→filter→load→instantiate path. It then registers
each provider with the container and tracks it for lifecycle so plugin
providers participate in the application lifecycle without belonging to
the orchestrator’s static list. Framework packages that register under
EP_PROVIDERS (admin, auth, sql, web, …) are not plugins and are
never auto-registered; they are composed explicitly by the application’s
module graph.
Attributes:
discovered_providers: Providers instantiated during register(),
used for lifecycle and by tests.
def __init__(state_path: str | Path | None = None) -> None
async def register(container: ContainerRegistrarProtocol) -> None
Discover and register enabled plugin providers, skipping the disabled set.
validate_plan enforces descriptor requires/conflicts
(advisory): plugins whose dependencies are missing or whose conflicts
are enabled are excluded from registration with a logged warning —
never raised, and boot is never blocked.
| Parameter | Type | Description |
|---|---|---|
container | ContainerRegistrarProtocol | The DI container registrar. |
async def boot(container: BootContainerProtocol) -> None
Boot all discovered providers.
| Parameter | Type | Description |
|---|---|---|
container | BootContainerProtocol | The DI container for boot phase. |
async def shutdown() -> None
Shut down discovered providers in reverse discovery order.
PluginsModule
Section titled “PluginsModule”Module that activates entry-point plugin discovery at boot.
Usage
# Option A — module auto-discovery in app config# discovery:# auto_discover: true (scans ``quadkit.modules``)## Option B — explicitfrom quadkit.plugins.module import PluginsModule
app.add_module(PluginsModule.configure())# Option A — module auto-discovery in app config# discovery:# auto_discover: true (scans ``quadkit.modules``)## Option B — explicitfrom quadkit.plugins.module import PluginsModule
app.add_module(PluginsModule.configure())def configure( cls, state_path: str | Path | None = None ) -> DynamicModule
Return a DynamicModule wiring PluginEngineProvider.
| Parameter | Type | Description |
|---|---|---|
state_path | str | Path | None | Optional explicit plugin-state file path. Defaults to the state module's env-var / default resolution. |
| Type | Description |
|---|---|
| DynamicModule | A DynamicModule exporting the plugin engine provider. |
Provider
Section titled “Provider”Base class for all Quadkit providers.
Providers are part of the composition root. They register bindings and wire the object graph during boot.
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
property state() -> ProviderState
property is_booted() -> bool
def reset_state() -> None
Reset the lifecycle state so this provider can be booted again.
Intended for test harnesses that restart an application while
reusing the same provider instances (e.g.
TestEnvironment.setup() after incremental provider
registration).
Note
Only the lifecycle state is reset; configuration and already registered services are left untouched.
property config() -> Any
def config(value: Any) -> None
Allow providers to assign configuration during initialization.
def from_config( cls, config: Any, **context: Any ) -> Self
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.
| 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). |
| Type | Description |
|---|---|
| Self | A new provider instance configured from *config*. |
async def register(container: ContainerRegistrarProtocol) -> None
Bind services into the container.
async def boot(container: BootContainerProtocol) -> None
Initialize services and wire dependencies.
async def shutdown() -> None
Tear down resources.
async def on_error( error: Exception, phase: str ) -> None
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.
| Parameter | Type | Description |
|---|---|---|
error | Exception | The exception that was raised. |
phase | str | Either 'boot' or 'shutdown' indicating which phase failed. |
async def health_check(timeout: float = 5.0) -> HealthCheckResult
Return health status of this provider.
ProviderPriority
Section titled “ProviderPriority”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
Section titled “ProviderState”Provider lifecycle state.
Registry
Section titled “Registry”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
Section titled “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.
Construction
Section titled “Construction”A registry instance is always empty — __init__ never registers
anything. There are exactly two ways to obtain a populated instance:
1. Explicit population (plugin/contributor registries) — entries are
supplied by outside code through register() / register_many()
registry = MyPluginRegistry()registry.register_many(discover_plugins()) # entry points, loadersregistry = MyPluginRegistry()registry.register_many(discover_plugins()) # entry points, loaders2. with_defaults() (in-package built-in registries) — when every
entry is a complete, static set shipped by the same package
registry = MyBuiltinRegistry.with_defaults()registry = MyBuiltinRegistry.with_defaults()Both idioms are interpreted as a single rule: with_defaults() returns
exactly the complete in-package built-in set declared by
_default_entries, while plugin registries declare no built-in set
and are populated by their loader. Calling with_defaults() on a
registry without a built-in set raises NotImplementedError rather
than silently returning a partial registry.
def __init__( name: str | None = None, *, allow_overwrite: bool = False, priority_key: Callable[[V], int] | None = None ) -> None
def with_defaults(cls) -> Self
Create a registry pre-populated with the class’s built-in set.
Returns exactly what _default_entries declares. Registries
whose entries are contributed externally (plugins, entry points) do
not override _default_entries; calling with_defaults() on
them raises NotImplementedError so the registry is never silently
partial — populate those explicitly with register().
| Type | Description |
|---|---|
| Self | A new, empty-at-init registry containing every declared default. |
| Exception | Description |
|---|---|
| NotImplementedError | When the class declares no built-in set. |
property name() -> str
Registry name for identification.
property allow_overwrite() -> bool
Whether overwriting existing keys is allowed.
def freeze() -> None
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.
property is_frozen() -> bool
Return whether this registry has been frozen.
def register( key: K, value: V, *, allow_overwrite: bool | None = None ) -> V
def register( key: K, value: None = None, *, allow_overwrite: bool | None = None ) -> Callable[[V], V]
def register( key: K, value: V | None = None, *, allow_overwrite: bool | None = None ) -> V | Callable[[V], V] | None
Register an item with the given key or use as decorator.
def register_factory( key: K, factory: Callable[[], V], *, allow_overwrite: bool | None = None ) -> None
Register a factory function for lazy instantiation.
def register_many(entries: Iterable[tuple[K, V]]) -> None
Register multiple (key, value) pairs in one call.
Use for plugin/contributor population: loaders collect entries
outside the registry and hand them over in bulk. Duplicate keys
follow the same rules as register (raises unless
allow_overwrite).
| Parameter | Type | Description |
|---|---|---|
entries | Iterable[tuple[K, V]] | Iterable of `(key, value)` pairs. |
def unregister(key: K) -> V | None
Unregister and return an item.
def get(key: K) -> V | None
def get( key: K, default: V ) -> V
def get( key: K, default: V | None = None ) -> V | None
Retrieve an item by key.
def resolve(key: K) -> V
Resolve an item, raising if not found.
def get_or_raise(key: K) -> V
Alias for resolve.
def has(key: K) -> bool
Check if key is registered (item or factory).
def keys() -> Iterable[K]
Return all registered keys (items only).
def values() -> Iterable[V]
Return all registered values.
def values_ordered() -> list[V]
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.
| Type | Description |
|---|---|
| list[V] | List of values sorted by priority, or insertion order if no priority_key is configured. |
def items() -> Iterable[tuple[K, V]]
Return all registered items.
def factories() -> Iterable[K]
Return keys with registered factories.
def all_keys() -> set[K]
Return all keys (items + factories).
def clear() -> None
Clear all registered items and factories.
def register_decorator( key: K | None = None, *, allow_overwrite: bool | None = None ) -> Callable[[V], V]
Decorator for easy registration.
RequestContext
Section titled “RequestContext”Context manager that sets request-scoped variables and restores them on exit.
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
Result
Section titled “Result”Base Result type. Not abstract — Ok and Err are the only variants.
def is_ok() -> bool
def is_err() -> bool
def unwrap() -> T
def unwrap_err() -> E
def unwrap_or(default: T) -> T
def unwrap_or_else(op: Callable[[E], T]) -> T
def map_sync(op: Callable[[T], U]) -> Result[U, E]
def map_err(op: Callable[[E], F]) -> Result[T, F]
def expect(message: str) -> T
def match( ok: Callable[[T], U], err: Callable[[E], U] ) -> U
async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]
def flatten() -> Result[Any, E]
def filter( predicate: Callable[[T], bool], error: E ) -> Result[T, E]
def ok_or(default: U) -> T | U
def from_exception( cls, exc: Exception, ok_type: type[T] = type(None) ) -> Result[T, Exception]
Wrap a caught exception into an Err result.
def to_optional() -> T | None
def inspect(op: Callable[[T], None]) -> Result[T, E]
def inspect_err(op: Callable[[E], None]) -> Result[T, E]
ResultPipeline
Section titled “ResultPipeline”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.
def __init__(initial: Result[T, E]) -> None
def then(func: Callable[[T], Result[U, E]]) -> ResultPipeline[U, E]
Chain a function that returns a Result, preserving the error type.
def map(func: Callable[[T], U]) -> ResultPipeline[U, E]
Apply a function to the value if Ok, preserving the error type.
def recover(func: Callable[[E], Result[T, F]]) -> ResultPipeline[T, F]
Recover from an error by applying a function that may produce a new error type.
def finalize() -> Result[T, E]
Get the final typed Result.
RetryOptions
Section titled “RetryOptions”Configuration for the retry operator.
Attributes:
max_attempts: Maximum number of subscription attempts. Defaults to 3.
delay: Base delay before retrying, in seconds. Defaults to 0.
backoff: "fixed" or "exponential". Defaults to "exponential".
max_delay: Upper bound for exponential backoff. Defaults to 30.
should_retry: Optional predicate; retry only when it returns True.
clock: Optional time source for backoff waits.
Request-scoped resolution context.
Caches scoped service instances for the lifetime of the scope. Singletons delegate to the root container. Scoped and transient constructors resolve their dependencies through this scope, never the root resolver.
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 scopeasync with container.create_scope() as scope: session = await scope.resolve(DbSession) # same instance within this scopeNested 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)async def resolve(service_type: type[T]) -> T
Resolve a service from this scope.
K-02: Falls back to parent scope if not found in current scope.
def create_scope() -> Scope
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.
async def call( func: Callable[Ellipsis, Awaitable[T] | T], *args: Any, **kwargs: Any ) -> T
Call a function with dependency injection (Protocol implementation).
def has(service_type: object) -> bool
Check if a service is registered (delegates to root container).
def can_resolve(service_type: object) -> bool
Internal injector capability; resolution itself stays scope-owned.
async def resolve_optional(service_type: type[T]) -> T | None
Resolve a service, returning None if not registered.
async def resolve_all(service_type: type[T]) -> list[T]
Resolve all registered implementations that are subtypes of a service type.
async def dispose() -> None
Dispose all scoped resources.
SerializationError
Section titled “SerializationError”JSON serialisation or deserialisation failed.
Raised by:
JsonSerializerimplementations- Domain model serialisation
- Schema validation during decode
Stream
Section titled “Stream”A cold stream backed by an async iterator.
Example
async def gen(): yield 1 yield 2
stream: EventStream[int] = Stream(gen())assert [item async for item in stream] == [1, 2]Note
Single-pass: this wraps one AsyncIterator and does not restart
it. A second full iteration of the same Stream instance yields
nothing — construct a fresh Stream (or call the producing
function again) to consume the same logical source twice.
def __init__(source: AsyncIterator[T]) -> None
Initialize a stream from an async iterator.
| Parameter | Type | Description |
|---|---|---|
source | AsyncIterator[T] | Any async iterator (generator, another stream, channel). |
def pipe(*operators: Op[Any, Any]) -> EventStream[Any]
Apply operators left to right.
| Parameter | Type | Description |
|---|---|---|
operators | Op[Any, Any] | Piped operators; each transforms one stream into another. |
| Type | Description |
|---|---|
| EventStream[Any] | The composed stream. |
Subject
Section titled “Subject”A hot, multicast event source.
Each subscriber iterates a private bounded channel. Publishing fans
out to every subscriber; on_overflow decides what happens when a
subscriber’s channel is full:
"block"(default): suspend the producer until there is room."drop_oldest": evict the subscriber’s oldest queued item and deliver the new one — a slow consumer loses backlog, not the live feed (the right policy for dashboards and monitors)."drop_latest": discard the incoming item for that subscriber, keeping everything it already has queued.
replay=N keeps the newest N published items in a ring buffer
and seeds every new subscriber with them, so late joiners see recent
history before the live feed — the pattern reconnecting SSE clients
need. Registration is synchronous, so an item published while a
subscriber is being set up is delivered exactly once: via replay or
via the channel, never both and never neither.
Example
subject = Subject[int](replay=10, on_overflow="drop_oldest")
async for item in subject: print(item) # subscriber (replays up to 10 items)
await subject.publish(1) # producerdef __init__( channel_capacity: int = 256, on_overflow: str = 'block', replay: int = 0 ) -> None
Initialize a subject.
| Parameter | Type | Description |
|---|---|---|
channel_capacity | int | Per-subscriber buffer size. `0` is unbounded. |
on_overflow | str | `"block", "drop_oldest" or "drop_latest"`; see the class docstring for the policies. |
replay | int | Number of recent items buffered for late subscribers. `0` disables replay. |
| Exception | Description |
|---|---|
| ValueError | If `on_overflow` is not a known mode. |
property subscriber_count() -> int
Number of subscribers currently attached to the fan-out.
def replay_snapshot() -> list[T_subject]
Copy of the replay buffer, oldest item first.
async def publish(item: T_subject) -> None
Fan out an item to all subscribers.
The item is appended to the replay buffer (if any) before the
fan-out, so a subscriber registering mid-publish sees it exactly
once. With a drop mode the fan-out never suspends, which keeps
publish atomic with respect to the event loop.
| Parameter | Type | Description |
|---|---|---|
item | T_subject | Item to publish. |
| Exception | Description |
|---|---|
| BackpressureError | With `on_overflow="drop_oldest"` when a subscriber channel is somehow still full after eviction. |
async def complete() -> None
Close all subscriber channels; remaining buffered items drain.
async def error(exc: BaseException) -> None
Terminate all subscribers by raising exc at their next item.
| Parameter | Type | Description |
|---|---|---|
exc | BaseException | The exception consumers will observe. |
Note
Publishes after error() are ignored, mirroring complete().
def subscribe() -> Subscription[T_subject]
Register a subscriber and return its handle.
Registration is synchronous (no awaits), so no publish can slip between seeding the replay history and joining the live fan-out.
| Type | Description |
|---|---|
| Subscription[T_subject] | A Subscription yielding replayed history followed by live items. Close it early with `await subscription.aclose()`. |
def pipe(*operators: Op[Any, Any]) -> EventStream[Any]
Apply operators left to right to this subject’s stream.
| Parameter | Type | Description |
|---|---|---|
operators | Op[Any, Any] | Piped operators; each transforms one stream into another. |
| Type | Description |
|---|---|
| EventStream[Any] | The composed stream. |
Subscription
Section titled “Subscription”A registered subscriber's handle over its private channel.
Iterate with async for to consume replayed history followed by
live items. When done consuming early (before complete()/
error() end the stream), call aclose() — it detaches the
channel from the subject’s fan-out so the publisher stops buffering
for a consumer that is gone. Abandoned subscriptions are detached by
the garbage collector as a safety net.
def __init__( channel: BoundedChannel[T_subject], on_done: Callable[[], None] ) -> None
async def aclose() -> None
Detach from the subject’s fan-out; idempotent.
The channel itself stays open — it belongs to the subject’s
lifecycle — but it no longer receives published items and the
subject’s subscriber_count drops immediately.
Functions
Section titled “Functions”SystemClock
Section titled “SystemClock”
def SystemClock() -> ClockProtocol
as_result
Section titled “as_result”
def as_result(*exceptions: type[Exception]) -> Callable[[Callable[P, Awaitable[T]]], Callable[P, Awaitable[Result[T, Exception]]]]
For sync functions use as_result_sync.
At least one exception type must be specified to prevent silently wrapping unexpected infrastructure failures.
| Parameter | Type | Description |
|---|
| 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
Section titled “as_result_sync”
def as_result_sync(*exceptions: type[Exception]) -> Callable[[Callable[P, T]], Callable[P, Result[T, Exception]]]
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.
| Parameter | Type | Description |
|---|
| 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
Section titled “buildable”
def buildable(cls: type[T]) -> type[T]
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
@buildableclass 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_age(30) .build())@buildableclass 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("[email protected]") .with_age(30) .build())| Parameter | Type | Description |
|---|---|---|
cls | type[T] | The class to decorate. Must have a standard `__init__` with type-annotated parameters. |
| Type | Description |
|---|---|
| type[T] | The same class with an attached `.builder()` static method. |
builder_field
Section titled “builder_field”
def builder_field(field_or_func: str | Callable[Ellipsis, Any]) -> Callable[Ellipsis, Any]
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
Section titled “collect”Collect an iterable of Results into a Result of a list.
create_app
Section titled “create_app”
def create_app( name: str = 'quadkit-app', config: QuadkitConfig | None = None ) -> 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
quadkit.plugins.discovery.discover_providers — the plugin system
ships inside the core quadkit package.
| Parameter | Type | Description |
|---|---|---|
name | str | Human-readable application name used in log output. Defaults to `"quadkit-app"`. |
config | QuadkitConfig | None | Application configuration. If omitted an empty QuadkitConfig is created, which reads values from environment variables via the standard `LX_*` prefix. |
| Type | Description |
|---|---|
| Application | A new Application ready for provider registration and boot. |
Example
app = create_app(config=QuadkitConfig.from_env_profile("production"))app.add_provider(WebProvider(web_config))await app.boot()app = create_app(config=QuadkitConfig.from_env_profile("production"))app.add_provider(WebProvider(web_config))await app.boot()create_module
Section titled “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
discover_plugins
Section titled “discover_plugins”
def discover_plugins() -> list[PluginDescriptor]
Metadata-only — never imports the plugin’s actual Provider class,
so this is cheap to call even for listing purposes (e.g. an admin UI).
discover_providers
Section titled “discover_providers”
def discover_providers(disabled: set[str] | None = None) -> list[Provider]
Only entry points referenced by an installed PluginDescriptor
(:group:EP_PLUGINS) are considered — framework packages that happen
to register under EP_PROVIDERS (admin, auth, sql, web, …) are
not plugins and are left out; they are composed explicitly by the
application’s module graph instead.
| Parameter | Type | Description |
|---|---|---|
disabled | set[str] | None | Entry-point names to skip (e.g. from quadkit.plugins.state.load_disabled). |
| Type | Description |
|---|---|
| list[Provider] | Instantiated `Provider objects for every enabled, constructible plugin entry point. Entries that fail to load, aren't a Provider` subclass, or require constructor arguments are skipped with a log. |
inject
Section titled “inject”
def inject(func_or_cls: Any) -> Any
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().
| Parameter | Type | Description |
|---|---|---|
func_or_cls | Any | The class or async function to decorate. |
| Type | Description |
|---|---|
| Any | The decorated class or function with automatic DI support. |
| Exception | Description |
|---|---|
| TypeError | If applied to a synchronous (non-async) function. |
injectable
Section titled “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]
The scope metadata is honored where a descriptor is created for the
class: by discover_injectables() (module scanning) and by explicit
Provider registration. Resolving the class directly from a bare
Container without either step is not supported — the container only
auto-wires classes carrying @inject-style constructor markers and
raises QK_ERR_DI_004 otherwise.
| 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). |
| Type | Description |
|---|---|
| Callable[[type[T]], type[T]] | type[T] | The decorator function or the decorated class. |
load_disabled
Section titled “load_disabled”
def load_disabled(path: str | Path | None = None) -> set[str]
Missing or corrupt state never prevents application boot — a corrupt
file is preserved as <name>.corrupt-<ts> and reading returns an
empty set. Legacy unversioned files ({"disabled": [...]}) load fine.
| Parameter | Type | Description |
|---|---|---|
path | str | Path | None | Explicit state-file path. Falls back to the `QUADKIT_PLUGINS_STATE_PATH env var, then .quadkit/plugins.json` relative to the working directory. |
| Type | Description |
|---|---|
| set[str] | Set of disabled entry-point names (possibly empty). |
module
Section titled “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]
partition
Section titled “partition”
def partition(results: Iterable[Result[T, E]]) -> tuple[list[T], list[E]]
provide
Section titled “provide”
def provide(func: Callable[Ellipsis, Any]) -> type[FunctionProvider]
refresh_namespace
Section titled “refresh_namespace”
def refresh_namespace() -> None
Mutated in place: children such as quadkit.ai hold a reference to the
same list and notice the change. Idempotent; call it after a path entry
gains a tree.
request_context
Section titled “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]
with request_scope(registry, request_id=“abc”) as req_ctx: …
def retry(options: RetryOptions | None = None) -> Any
| Parameter | Type | Description |
|---|---|---|
options | RetryOptions | None | Retry configuration; defaults to `RetryOptions()`. |
| Type | Description |
|---|---|
| Any | An operator that retries the piped source. |
Note
Only the successful attempt’s output is emitted; failed attempts
are discarded. The source must be re-subscribable (replayable) for
retries to take effect — single-pass Stream sources are not.
run_application
Section titled “run_application”
async def run_application(app: Application) -> None
Starts the application, waits for SIGINT/SIGTERM, then performs a
graceful shutdown. Handles :exc:KeyboardInterrupt and
:exc:SystemExit silently so the process terminates cleanly.
| Parameter | Type | Description |
|---|---|---|
app | Application | The Application instance to run. |
save_disabled
Section titled “save_disabled”
def save_disabled( names: set[str], path: str | Path | None = None ) -> None
Atomic (temp + fsync + os.replace) and serialized via flock
so concurrent writers cannot interleave read-modify-write cycles.
| Parameter | Type | Description |
|---|---|---|
names | set[str] | Set of disabled plugin entry-point names. |
path | str | Path | None | Explicit state-file path (env var / default fallback applies when omitted). |
| Exception | Description |
|---|---|
| PluginStateError | If the state cannot be written (disk full, permission denied, etc.). |
scoped
Section titled “scoped”
def scoped(cls: type[T]) -> type[T]
A scoped service creates one instance per scope (e.g., per request).
| Parameter | Type | Description |
|---|---|---|
cls | type[T] | The class to register as scoped. |
| Type | Description |
|---|---|
| type[T] | The decorated class marked for scoped registration. |
def share( source: EventStream[Any], channel_capacity: int = 256, on_overflow: str = 'block', replay: int = 0 ) -> Subject[Any]
| Parameter | Type | Description |
|---|---|---|
source | EventStream[Any] | Cold or hot source stream. |
channel_capacity | int | Per-subscriber buffer size. |
on_overflow | str | Overflow policy; see Subject. |
replay | int | Number of recent items buffered for late subscribers. |
| Type | Description |
|---|---|
| Subject[Any] | A Subject fed by the source. Iterating it subscribes to the live feed. |
Note
If the pump task fails, subscribers observe the pump’s exception at
their next item (recover upstream with ops.catch); a cancelled
pump completes the subject cleanly.
singleton
Section titled “singleton”
def singleton(cls: type[T]) -> type[T]
A singleton service is created once and shared across all resolutions.
| Parameter | Type | Description |
|---|---|---|
cls | type[T] | The class to register as a singleton. |
| Type | Description |
|---|---|
| type[T] | The decorated class marked for singleton registration. |
transient
Section titled “transient”
def transient(cls: type[T]) -> type[T]
A transient service creates a new instance each time it is resolved.
| Parameter | Type | Description |
|---|---|---|
cls | type[T] | The class to register as transient. |
| Type | Description |
|---|---|
| type[T] | The decorated class marked for transient registration. |
try_catch
Section titled “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]]
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.
| 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
Section titled “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]
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).
| 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")update_disabled
Section titled “update_disabled”
def update_disabled( mutator: Any, path: str | Path | None = None ) -> set[str]
Load, mutate, and save happen inside a single flock critical
section, so concurrent readers/writers (e.g. two admin sessions in
different processes) cannot lose updates.
| Parameter | Type | Description |
|---|---|---|
mutator | Any | Callable mapping the current disabled set to the new disabled set, e.g. `lambda current: current | {"rag"}`. |
path | str | Path | None | Explicit state-file path (env var / default fallback applies when omitted). |
| Type | Description |
|---|---|
| set[str] | The resulting disabled set after `mutator` was applied. |
| Exception | Description |
|---|---|
| PluginStateError | If the state cannot be read or written. |
Exceptions
Section titled “Exceptions”ContainerError
Section titled “ContainerError”Base container error.
def __init__( message: str = 'Container error', **kwargs: Any ) -> None
DomainError
Section titled “DomainError”Business/domain-level errors (validation failures, not found, auth).
def __init__( message: str = 'Domain error', **kwargs: Any ) -> None
InfrastructureError
Section titled “InfrastructureError”System-level failures (DB down, network timeouts).
def __init__( message: str = 'Infrastructure error', **kwargs: Any ) -> None
ModuleError
Section titled “ModuleError”Module configuration or lifecycle error.
def __init__( message: str = 'Module error', **kwargs: Any ) -> None
NotFoundError
Section titled “NotFoundError”Resource not found error (domain-level).
def __init__( message: str = 'Not found', **kwargs: Any ) -> None
PluginStateError
Section titled “PluginStateError”Raised when plugin enable/disable state cannot be persisted or read.
The admin surface converts this into a flash message and audit entry — a storage failure must never surface as an unhandled 500.
The optional path argument is accepted for compatibility but is
intentionally discarded so local filesystem paths are never retained on
the exception.
def __init__( message: str, *, path: str | None = None ) -> None
ProviderError
Section titled “ProviderError”Provider configuration or lifecycle error.
def __init__( message: str = 'Provider error', **kwargs: Any ) -> None
QuadkitError
Section titled “QuadkitError”Root exception for the entire Quadkit ecosystem.
All Quadkit 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., ‘QK_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
def __init__( message: str | None = None, details: dict[str, Any] | None = None, cause: BaseException | None = None, hint: str | None = None, **kwargs: Any ) -> None
property docs_url() -> str
Return the documentation URL for this error code.
def to_dict() -> dict[str, Any]
Serialize exception to dictionary for logging/API responses.
def with_details(**kwargs: Any) -> Self
Return self with additional detail key-value pairs merged in.
Mutates and returns self to preserve the exception subtype.
def with_hint(hint: str) -> Self
Return self with a human-readable hint attached.
def with_cause(cause: BaseException) -> Self
Return self with __cause__ set to cause.
def format() -> str
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.
QuadkitException
Section titled “QuadkitException”Base exception for all quadkit domain errors.
Extends QuadkitError so
that isinstance(exc, QuadkitError) 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 QuadkitException as e: logger.error("resolution_failed", error=str(e)) raisetry: service = await container.resolve(MyService)except QuadkitException as e: logger.error("resolution_failed", error=str(e)) raise