Patterns & Recipes — Aquilia Documentation
Comprehensive guide and documentation for Patterns & Recipes in the Aquilia framework. View API reference, examples, and implementation patterns.
Dependency Injection / Patterns & Recipes Patterns & Real-World Recipes A cookbook of production-tested DI patterns — from basic registration to large-scale application architecture. Every recipe is runnable and reflects the actual aquilia.di implementation. 1. Basic Service Registration The idiomatic path is declarative: decorate a class with @service and list it in your module's manifest.py. The runtime discovers it, builds a ClassProvider, and wires its constructor. Scope defaults to "app". from aquilia.di import service @service() # scope="app" (one instance per app container) class GreetingService: def greet(self, name: str) -> str: return f"Hello, !" from aquilia.aquilary import AppManifest manifest = AppManifest( name="users", services=["modules.users.services:GreetingService"], controllers=["modules.users.controllers:UsersController"], ) For programmatic setup (tests, scripts) register directly against a container: from aquilia.di import Container, ClassProvider container = Container(scope="app") container.register(ClassProvider(GreetingService, scope="app")) svc = await container.resolve_async(GreetingService) print(svc.greet("Ada")) # Hello, Ada! 2. Constructor Injection Aquilia reads type hints from __init__ and resolves each annotated parameter. No annotation and no default raises DIError at container build; a default makes the parameter optional (skipped by DI). @service() class UserRepository: def __init__(self, db: Database): # resolved by type self.db = db @service() class UserService: def __init__(self, repo: UserRepository, cache: CacheBackend): self.repo = repo self.cache = cache async def get(self, user_id: str): if cached := await self.cache.get(f"user: "): return cached user = await self.repo.find(user_id) await self.cache.set(f"user: ", user, ttl=300) return user Async construction: if a class defines an async def async_init(self) method, the ClassProvider calls it automatically after __init__ — useful when setup needs await (connect a pool, warm a cache). @service(scope="app") class SearchIndex: def __init__(self, config: AppConfig): self.config = config self.client = None async def async_init(self): self.client = await connect_elastic(self.config.es_url) 3. Factory Providers When construction needs logic — reading config, choosing an implementation, calling an async connector — use a factory. The factory's own parameters are injected too. Use @provides(Token) when the return type is abstract. from aquilia.di import factory, provides, FactoryProvider # Simple factory — token is the function @factory(scope="singleton") async def create_http_client(config: AppConfig) -> HttpClient: return HttpClient(base_url=config.api_url, timeout=config.timeout) # @provides — bind an abstract token to a concrete build @provides(PaymentGateway, scope="app", tag="live") def build_gateway(config: AppConfig) -> PaymentGateway: return StripeGateway(config.stripe_key) # Programmatic equivalent container.register(FactoryProvider(create_http_client, scope="singleton")) 4. Configuration Service A config object is the classic singleton: build it once, inject it everywhere. Bind a ready-made instance with ValueProvider (or register_instance) so DI hands back the same object without instantiating anything. from aquilia.di import ValueProvider @dataclass class AppConfig: db_url: str api_url: str timeout: float = 30.0 config = AppConfig(db_url=env("DATABASE_URL"), api_url=env("API_URL")) # Bind the concrete instance under the AppConfig token container.register(ValueProvider(value=config, token=AppConfig, scope="singleton")) # Now every service that asks for AppConfig gets this exact object @service(scope="app") class Mailer: def __init__(self, config: AppConfig): self.config = config 5. Repository & Database Pattern Bind an abstract repository interface to a concrete backend with container.bind(). Consumers depend on the interface, so you can swap SQL for in-memory in tests without touching call sites. from abc import ABC, abstractmethod class UserRepo(ABC): @abstractmethod async def find(self, user_id: str) -> User | None: ... class SqlUserRepo(UserRepo): def __init__(self, db: Database): self.db = db async def find(self, user_id: str): return await self.db.fetch_one( "SELECT * FROM users WHERE id = $1", user_id ) # parameterized — never string-format SQL # Bind interface to implementation (creates a ClassProvider internally) container.bind(UserRepo, SqlUserRepo, scope="app") @service() class ProfileService: def __init__(self, users: UserRepo): # gets SqlUserRepo self.users = users 6. Tagged Providers (Multiple Implementations) When several providers satisfy one token, disambiguate with a tag. Register each under a tag, then select one at the injection site with Inject(tag=...). Resolving an ambiguous token without a tag surfaces the mismatch. from typing import Annotated from aquilia.di import Inject, ClassProvider container.register(ClassProvider(RedisCache, scope="app"), tag="redis") container.register(ClassProvider(MemoryCache, scope="app"), tag="memory") @service() class SessionStore: def __init__( self, hot: Annotated[CacheBackend, Inject(tag="redis")], cold: Annotated[CacheBackend, Inject(tag="memory")], ): self.hot = hot self.cold = cold # Direct resolution with a tag redis = await container.resolve_async(CacheBackend, tag="redis") 7. Optional Dependencies Two ways to make a dependency optional. Both resolve to None when the provider is absent instead of raising ProviderNotFoundError: from typing import Annotated, Optional from aquilia.di import Inject @service() class AnalyticsService: def __init__( self, # (a) Optional[T] type — DI marks it optional automatically tracer: Optional[Tracer], # (b) explicit Inject(optional=True) metrics: Annotated[MetricsClient, Inject(optional=True)], ): self.tracer = tracer # None if no Tracer registered self.metrics = metrics # None if no MetricsClient registered async def record(self, event: str): if self.metrics: # guard — may be None await self.metrics.incr(event) A constructor parameter with a default value is also treated as optional and skipped by DI if unresolved. 8. Lazy Resolution (Breaking Cycles) Two singletons that need each other form a cycle the graph validator rejects. A LazyProxyProvider defers resolution until first attribute access, breaking the construction-time loop. Reach for it only when a redesign (extract an interface, use events) isn't practical. from aquilia.di import LazyProxyProvider, ClassProvider # ServiceA needs ServiceB and vice-versa. container.register(ClassProvider(ServiceB, scope="app")) # Register a lazy proxy for B under a distinct token A depends on: container.register(LazyProxyProvider(token=LazyB, target_token=ServiceB)) @service() class ServiceA: def __init__(self, b: LazyB): # gets a proxy, not a live ServiceB self._b = b # ServiceB is built on first b. access The proxy resolves synchronously on first access via a persistent per-thread event loop. It refuses to resolve inside a running event loop (would deadlock) — so the first touch must happen outside the async hot path, or you should resolve ServiceB explicitly with await. 9. Request-Scoped Services A request-scoped service lives for one HTTP request and is cached in the per-request child container. Perfect for a unit-of-work, a request-bound logger, or the current user. The ASGI layer creates and disposes the request container automatically. @service(scope="request") class UnitOfWork: def __init__(self, db: Database): # db is app-scoped, shared self.db = db self.tx = None async def async_init(self): self.tx = await self.db.begin() async def commit(self): await self.tx.commit() @service(scope="request") class OrderService: def __init__(self, uow: UnitOfWork): # same UoW for the whole request self.uow = uow Captive dependency: a singleton/app service must not inject a request-scoped one — the short-lived instance would be captured for the process lifetime. With scope_enforcement="raise" this throws ScopeViolationError at resolution. 10. Transient & Pooled Services transient builds a fresh instance on every resolve (stateless helpers, per-use builders). pooled hands out reusable instances from a bounded asyncio.Queue (heavy clients, capped concurrency). from aquilia.di import PoolProvider @service(scope="transient") class RequestIdGenerator: # new one every time it is injected def next(self) -> str: return uuid4().hex # Pool of 10 heavy clients; fast-fail if 256 callers pile up async def make_worker() -> HeavyWorker: return await HeavyWorker.connect() container.register(PoolProvider( make_worker, max_size=10, token=HeavyWorker, acquire_timeout=30.0, max_waiters=256, )) # Auto acquire/release async with pool.acquire() as worker: await worker.run(job) 11. Service Composition Compose small, single-responsibility services into higher-level ones. DI resolves the whole tree in dependency order, deduplicating shared leaves (a Database depended on by three services is built once). @service() class NotificationService: def __init__(self, mailer: Mailer, sms: SmsClient): self.mailer, self.sms = mailer, sms @service() class CheckoutService: def __init__( self, orders: OrderRepository, payments: PaymentGateway, notify: NotificationService, ): self.orders, self.payments, self.notify = orders, payments, notify async def checkout(self, cart): order = await self.orders.create(cart) await self.payments.charge(order.total) await self.notify.mailer.send(order.receipt()) return order 12. Cross-App Dependencies (depends_on) In a multi-module app, one module can consume another's services — but the edge must be declared in the manifest's depends_on. The registry validates the edge statically; the runtime wires a dependency link so resolution falls through to the owning app's container. manifest = AppManifest( name="billing", depends_on=["auth"], # billing may inject auth-owned services services=["modules.billing.services:InvoiceService"], ) @service() class InvoiceService: # AuthService is owned by the "auth" app; resolvable because # billing declares depends_on=["auth"]. def __init__(self, auth: AuthService): self.auth = auth An undeclared cross-app dependency raises CrossAppDependencyError at boot (static validation). A cycle between app links raises DependencyCycleError at resolution instead of deadlocking. The owning app instantiates and caches its singletons exactly once. 13. Plugin / Extension Scenario A DIPlugin hooks into registry construction — auto-register a family of providers, observe every registration, or inspect built containers. Ideal for a shared library that self-wires when installed. from aquilia.di import DIPlugin, register_plugin, ClassProvider class AuditPlugin(DIPlugin): name = "audit" # stable id — re-registering replaces def on_registry_build(self, registry): registry.add_provider(ClassProvider(AuditLogger, scope="app")) def on_container_built(self, container): log.info("DI container ready with audit wiring") register_plugin(AuditPlugin()) # honoured when enable_plugins=True Combine with a provider interceptor for around-advice on instantiation (timing, tracing) — see Advanced DI. 14. Testing & Mocking Swap real services for mocks without rebuilding the container. override_container is an async context manager that force-replaces a provider for the duration of a block and restores it on exit. import pytest from unittest.mock import AsyncMock from aquilia.di.testing import override_container @pytest.mark.asyncio async def test_checkout_charges_card(app_container): fake_gateway = AsyncMock() fake_gateway.charge.return_value = async with override_container(app_container, PaymentGateway, fake_gateway): checkout = await app_container.resolve_async(CheckoutService) await checkout.checkout(sample_cart) fake_gateway.charge.assert_awaited_once() For whole-suite wiring use TestRegistry (relaxed: cross-app checks off, cycles tolerated) with an overrides map, or the built-in di_container / request_container pytest fixtures. from aquilia.di import TestRegistry, MockProvider registry = TestRegistry.from_manifests( manifests, overrides= , ) container = registry.build_container() 15. Large Application Architecture At scale, lean on the manifest-first model and a few conventions: . ))} class ProdEnv(BaseEnv): class di(BaseEnv.di): scope_enforcement = "raise" # captive deps abort boot strict_service_registration = True # a bad service fails fast parallel_resolution = True # concurrent independent deps pool_max_waiters = 256 Advanced Errors & Troubleshooting )
Go to Homepage