layer

effectpy.layer

Layer

Composable resource builder for dependency injection.

A Layer describes how to build services into a Context and how to clean them up. Layers can be composed sequentially (+) for dependencies or in parallel (|) for independent services.

Parameters:

Name	Type	Description	Default
acquire	Callable[[Context, dict], Awaitable[Context]]	Function to build services into a context	required
release	Callable[[Context, dict], Awaitable[None]]	Function to clean up services	required

Example

# Define a database layer
def build_db(ctx: Context, memo: dict) -> Context:
    db = Database("postgresql://...")
    return ctx.with_service(Database, db)

def teardown_db(ctx: Context, memo: dict) -> None:
    db = ctx.get(Database)
    await db.close()

DatabaseLayer = Layer(build_db, teardown_db)

__add__(other)

Sequential layer composition - build dependencies first.

The left layer is built first, then the right layer is built with access to the left layer's services. Use this when the right layer depends on services from the left layer.

Parameters:

Name	Type	Description	Default
other	'Layer'	Layer to compose sequentially after this one	required

Returns:

Type	Description
'Layer'	New layer representing sequential composition

Example

# Database depends on Logger
AppLayer = LoggerLayer + DatabaseLayer

__or__(other)

Parallel layer composition - build services concurrently.

Both layers are built concurrently. Use this when the layers are independent and don't depend on each other's services.

Parameters:

Name	Type	Description	Default
other	'Layer'	Layer to compose in parallel with this one	required

Returns:

Type	Description
'Layer'	New layer representing parallel composition

Example

# Logger and Metrics are independent
ObservabilityLayer = LoggerLayer | MetricsLayer

build(parent) async

Build this layer's services into a new context.

Parameters:

Name	Type	Description	Default
parent	Context	The parent context to extend	required

Returns:

Type	Description
Context	New context with this layer's services added

build_scoped(parent, scope, memo=None) async

Build this layer and register cleanup with a scope.

The layer is built and a finalizer is added to the scope to ensure proper cleanup when the scope closes.

Parameters:

Name	Type	Description	Default
parent	Context	The parent context to extend	required
scope	Scope	The scope to register cleanup with	required
memo	dict | None	Optional memoization dictionary for complex layers	None

Returns:

Type	Description
Context	New context with this layer's services and guaranteed cleanup

Example

scope = Scope()
env = await DatabaseLayer.build_scoped(Context(), scope)
# Use database...
await scope.close()  # Database automatically cleaned up

from_resource(t, mk, close)

Create a Layer from simple build and teardown functions.

This is a convenient helper for creating layers when you have simple functions to create and destroy a resource.

Parameters:

Name	Type	Description	Default
t	type	The service type to provide	required
mk	Callable[[Context], Awaitable[Any]]	Function to create the service instance	required
close	Callable[[Any], Awaitable[None]]	Function to clean up the service instance	required

Returns:

Type	Description
Layer	A new Layer that manages the resource

Example

class Database:
    def __init__(self, url: str): ...
    async def close(self): ...

DatabaseLayer = from_resource(
    Database,
    lambda ctx: Database("postgresql://localhost:5432/db"),
    lambda db: db.close()
)