Advanced — Aquilia Documentation
Comprehensive guide and documentation for Advanced in the Aquilia framework. View API reference, examples, and implementation patterns.
Dependency Injection / Advanced Advanced DI & Testing Overrides Customize the resolution graph dynamically, swap providers at runtime during tests, and configure complex factory pipelines. Decorators Cheat Sheet Decorator Scope Description , , , , , , , ].map((row, i) => ( ))} DI Settings Every runtime knob for the container lives in one typed, immutable DISettings object. Configure it declaratively through the di section of your workspace.py — the server reads it at boot and calls configure_di() for you. from aquilia import AquilaConfig class BaseEnv(AquilaConfig): class di(AquilaConfig.DI): scope_enforcement = "warn" # "warn" | "raise" | "off" parallel_resolution = False class DevEnv(BaseEnv): class di(BaseEnv.di): diagnostics_enabled = True # trace every resolution in dev class ProdEnv(BaseEnv): class di(BaseEnv.di): scope_enforcement = "raise" # fail-fast on captive deps parallel_resolution = True # resolve independent deps concurrently pool_max_waiters = 256 # fast-fail an exhausted pool Setting Default Purpose ))} In tests or scripts you can configure the container directly. Invalid values raise DIConfigFault at construction, so bad config surfaces at boot rather than at first resolution: from aquilia.di import DISettings, configure_di, get_di_settings, reset_di_settings configure_di(DISettings(scope_enforcement="raise", parallel_resolution=True)) assert get_di_settings().strict_scopes is True # Test teardown — restore permissive defaults reset_di_settings() Provider Interceptors Interceptors wrap a provider's instantiation with around-advice (AOP) — logging, timing, tracing, caching — without touching the service class. Wrap any provider with intercept(). Interceptors run in registration order, first = outermost; call nxt() to proceed, or skip it to short-circuit with your own object. from aquilia.di import ProviderInterceptor, intercept, ClassProvider class TimingInterceptor(ProviderInterceptor): async def around_instantiate(self, ctx, nxt): import time start = time.perf_counter() obj = await nxt() # proceed to real instantiation elapsed = time.perf_counter() - start print(f"built in us") return obj # Wrap a provider — interceptors run first=outermost provider = intercept(ClassProvider(UserService, scope="app"), TimingInterceptor()) container.register(provider) intercept(P, A, B) yields the chain A(in) → B(in) → B(out) → A(out). The wrapped InterceptingProvider mirrors the inner provider's token, scope, and tags. Wrapping with an empty interceptor list raises DIFault (DI_NO_INTERCEPTORS). DI Plugins A DIPlugin hooks into registry construction to auto-register providers, observe registrations, or inspect built containers — ideal for cross-cutting concerns like auto-wiring repositories. Register once with register_plugin(); hooks fire during boot when enable_plugins is on (default). from aquilia.di import DIPlugin, register_plugin, ClassProvider class RepositoryPlugin(DIPlugin): name = "repository-autoreg" # stable id — re-registering replaces def on_registry_build(self, registry): # Runs after manifests load, before the graph is built registry.add_provider(ClassProvider(UserRepository, scope="app")) def on_provider_registered(self, container, provider): ... # fires per register() call def on_container_built(self, container): ... # fires once each app container is built register_plugin(RepositoryPlugin()) Failure-isolated: a plugin hook that raises is logged and skipped — it never crashes boot. Manage the registry with unregister_plugin(name), get_plugins(), and clear_plugins() (test teardown). Plugins are deduplicated by .name. Cross-App Links & Runtime Swaps When a module declares depends_on in its manifest, the runtime wires a dependency link between the two app containers via add_dependency_link(). A token missing locally (and up the parent chain) falls through to the linked sibling app; the owning container instantiates and caches its own singletons exactly once. Undeclared cross-app dependencies still raise ProviderNotFoundError, and link cycles raise DependencyCycleError instead of deadlocking. # Atomically replace a provider at runtime (copy-on-write safe, evicts the # cached instance). Distinct from the test-only override_container helper. await container.replace_provider(EmailService, ClassProvider(SmtpEmailService, scope="app")) # Generic hierarchical child (per-tenant trees, multi-level scopes) child = container.create_child(scope="app", own_lifecycle=True) TestRegistry Overrides Swap components with mock implementations during unit or integration testing: from aquilia.di import TestRegistry, MockProvider # Create a test registry delegating to production setup test_reg = TestRegistry(base=production_registry) # Override with custom value or MockProvider test_reg.override(EmailService, MockProvider( send=AsyncMock(return_value=True) )) # Override database connection pools with mock mocks test_reg.override(DatabasePool, value=FakeDbPool()) Pytest Fixtures Use the built-in context overrides helper to automatically mock out components during test execution: import pytest from aquilia.di.testing import override_container @pytest.mark.asyncio async def test_user_creation(client, app_container): # Override UserService inside the app DI container temporarily: mock_service = MagicMock() with override_container(app_container, ): response = await client.post("/users", json= ) assert response.status_code == 201 mock_service.create.assert_called_once() Scopes Models )
Go to Homepage