Layers & Compositions — Aquilia Documentation
Comprehensive guide and documentation for Layers & Compositions in the Aquilia framework. View API reference, examples, and implementation patterns.
SUBSYSTEM / LAYERS & COMPOSITIONS Layers & Compositions Manage complex capability graph initialization using Layer and EffectScope . Decouple construction from usage, sort dependencies topologically, and acquire resources safely outside pipelines. The Layer Architecture In large microservice applications, capability providers (like database connection pools or cache servers) have complex initialization graphs. A database provider might depend on a configuration provider, while a metrics service might require both configuration and database logging connections. Inspired by Effect-TS, the Layer system allows you to build modular constructors that declare explicit dependencies. The framework's dependency compiler sorts these layers topologically at startup, resolving, building, and registering providers automatically. Layer API Reference class Layer: def __init__(self, name: str, factory: Callable, deps: list[str] = [], scope: str = "app") Constructs a composable layer. factory is a callable that receives resolved dependencies as keyword arguments and returns a provider. scope dictates whether the provider has application or request lifetime. @staticmethod def merge(*layers: Layer) -> LayerComposition Combines multiple layers into a unified composition. The compiler automatically resolves inter-layer dependencies and orders initialization topologically. @staticmethod def provide(layer: Layer, *providers: Layer) -> LayerComposition Expresses dependency injections explicitly. Builds the list of providers first, then feeds their constructed values as dependencies into the target layer. LayerComposition API Merging or chaining layers creates a LayerComposition. It manages sorting and mount operations: async def build_all(self, initial_deps: dict | None = None) -> dict[str, Any] Builds all layers in computed topological order. Returns a dictionary mapping capability names to constructed provider instances. Raises FlowError if circular dependencies are detected. async def register_with(self, registry: EffectRegistry, initial_deps: dict | None = None) -> None Builds all layers and mounts the output providers directly into the active EffectRegistry registry. Manual Capability Management: EffectScope While pipelines acquire capabilities automatically based on annotations, you can manage them manually using the EffectScope context manager. This is ideal for background tasks, CLI commands, or scripts that require scoped resource access: class EffectScope: def __init__(self, registry: EffectRegistry, effect_names: list[str], *, context = None, modes = None) Async context manager. On enter (__aenter__), it calls acquire() on all listed providers. On exit (__aexit__), it releases them, tracking whether exceptions occurred to trigger rollbacks or commits automatically. Topological Bootstrap Walkthrough The following code defines configuration, database, and cache capability layers. The layers are merged, resolved topologically, registered, and accessed inside an EffectScope . from aquilia.flow import Layer, EffectScope from aquilia.effects import EffectRegistry, DBTxProvider, CacheProvider # 1. Define capability layers config_layer = Layer( name="Config", factory=lambda: ) db_layer = Layer( name="DBTx", # Receives constructed config layer output as keyword argument 'Config' factory=lambda Config: DBTxProvider(Config["db_url"]), deps=["Config"] ) cache_layer = Layer( name="Cache", # Receives config output as 'Config' factory=lambda Config: CacheProvider(Config["cache_host"]), deps=["Config"] ) # 2. Merge Layers (Sorted Topologically: Config -> DBTx -> Cache) app_composition = Layer.merge(db_layer, cache_layer, config_layer) # 3. Mount providers to registry registry = EffectRegistry() await app_composition.register_with(registry) # 4. Access resources manually via EffectScope async def process_jobs(): async with EffectScope(registry, ["DBTx", "Cache"]) as effects: db = effects["DBTx"] cache = effects["Cache"] # Safe transaction execution await db.execute("UPDATE stats SET runs = runs + 1") await cache.set("last_run", "now") Flow Context & Nodes Built-in Effects )
Go to Homepage