Container — Aquilia Documentation
Comprehensive guide and documentation for Container in the Aquilia framework. View API reference, examples, and implementation patterns.
Dependency Injection / Container DI Container The Container is the central state engine for resolved services. It manages provider lifecycle transitions, caches instances by scope, and delegates queries up hierarchical container chains. Internal Structure The Container uses __slots__ with 8 attributes for direct memory allocation, bypassing class dictionary lookups entirely: Slot Type Purpose ))} Creating Containers from aquilia.di.core import Container # Root app container (created by Registry.build_container() internally) container = Container(scope="app") # With explicit parent (for manual hierarchies) request_container = Container(scope="request", parent=container) # Preferred: use the factory method for request scoping request_container = container.create_request_scope() # → Creates child with shared _providers (by reference), fresh _cache, # and _NullLifecycle (no-op lifecycle for lightweight request containers) Note: In production web workflows, you almost never create containers manually. The Registry.build_container() method builds the root container, and the ASGI server middleware executes create_request_scope() on every incoming request automatically. API Reference register(provider, *, tag=None) Register a Provider instance. A genuine local re-registration of the same token+tag raises a DIFault, but a child container may shadow a provider inherited from its parent. Fires the on_provider_registered plugin hook. from aquilia.di import ClassProvider provider = ClassProvider(UserService, scope="request") container.register(provider) # With a tag for disambiguation container.register(redis_provider, tag="redis") bind(interface, implementation, *, scope="app", tag=None) Bind an interface type to a concrete implementation. Creates a ClassProvider internally. from abc import ABC, abstractmethod from aquilia.controller import Controller, get class IUserRepo(ABC): @abstractmethod async def find(self, id: str): ... class PostgresUserRepo(IUserRepo): def __init__(self, pool: DatabasePool): self.pool = pool async def find(self, id: str): return await self.pool.fetch_one("SELECT * FROM users WHERE id=$1", id) # Bind interface → implementation container.bind(IUserRepo, PostgresUserRepo, scope="app") # Web Controllers resolve this automatically via constructor injection: class UserController(Controller): prefix = "/users" def __init__(self, repo: IUserRepo): # Resolved to PostgresUserRepo self.repo = repo @get("/ ") async def get_user(self, ctx): user = await self.repo.find(ctx.request.params["id"]) return ctx.json(user) await resolve_async(token, *, tag=None, optional=False) Primary async resolution path. Accepts concrete types, string tokens (e.g. "modules.auth.services:CrossAppService"), Annotated[T, Inject(...)] aliases, or direct Inject(...) markers. Automatically unwraps descriptor metadata via _unwrap_token(). Optimized for <3µs cached lookups with O(1) cache check and parent container delegation. Pass optional=True to get None instead of ProviderNotFoundError when unregistered. # Standard type resolution user_svc = await container.resolve_async(UserService) # String tokenized cross-module resolution auth_svc = await container.resolve_async( Annotated[Any, Inject("modules.auth.services:CrossAppService")] ) # Tagged resolution redis = await container.resolve_async(CacheBackend, tag="redis") # Optional — None if not registered tracer = await container.resolve_async(Tracer, optional=True) resolve(token, *, tag=None, optional=False) Synchronous resolution for non-async call sites. Drives the async path on a persistent per-thread event loop. Raises DIResolutionFault if called from inside a running event loop — in async code, always use resolve_async. await register_instance(token, instance, scope="request", tag=None) Register a pre-built object (wraps it in a ValueProvider). Used for request-scoped objects created outside DI — the ASGI layer registers the current Request this way. Always replaces any existing entry for the token. session = await engine.open_session(request) await container.register_instance(Session, session, scope="request") is_registered(token, tag=None) Returns True if a provider is registered for the token (checks this container and its parent chain). create_request_scope() Create a lightweight child container for request-scoped isolation. The child shares the parent's providers but has isolated caches and finalizers. create_child(scope="app", *, own_lifecycle=True) Generic hierarchical child container (copy-on-write provider dict; parent singletons resolved once at the owning level). Use for per-tenant or multi-level scope trees. add_dependency_link(app_name, container) Runtime counterpart to a manifest's depends_on. When a token is missing locally and up the parent chain, resolution falls through to the linked sibling app container. Wired automatically by the runtime; undeclared cross-app deps still raise ProviderNotFoundError, and link cycles raise DependencyCycleError. await replace_provider(token, provider, *, tag=None) Production-safe atomic hot-swap of a provider (copy-on-write safe, evicts the cached instance). Distinct from the test-only override_container. Emits a REGISTRATION diagnostic event. shutdown() Runs inline Dep() generator teardowns (LIFO) first, then drains finalizers in LIFO order (clean up database connections or file handlers), runs lifecycle shutdown hooks, and clears the instance cache. Caching Behavior by Scope Scope Cached? Where Cached Behavior ))} DI Overview Providers )
Go to Homepage