Context & Scope API Reference

Resource management is critical for robust async applications. These modules provide the foundation:

• Context: Type-safe service container for dependency injection
• Scope: Resource lifecycle manager with guaranteed LIFO cleanup

Context API

effectpy.context

Context

Type-safe service container for dependency injection.

Context holds services that effects can access. Services are stored by type, providing compile-time safety and runtime efficiency.

Parameters:

Name	Type	Description	Default
values	Dict[type, Any] | None	Optional initial services dictionary	None

Example

# Create context with services
ctx = (Context()
       .with_service(Logger, Logger())
       .with_service(Database, Database("postgresql://...")))

# Access services in effects
logger = ctx.get(Logger)
db = ctx.get(Database)

add(t, v)

Add a service to the context.

Returns a new Context with the added service. The original context is unchanged (contexts are immutable).

Parameters:

Name	Type	Description	Default
t	type[A]	The type of the service	required
v	A	The service instance	required

Returns:

Type	Description
'Context'	New context containing the added service

Example

new_ctx = ctx.add(Logger, Logger())
# or use the more convenient alias:
new_ctx = ctx.with_service(Logger, Logger())

get(t)

Get a service from the context by type.

Parameters:

Name	Type	Description	Default
t	type[A]	The type of service to retrieve	required

Returns:

Type	Description
A	The service instance

Raises:

Type	Description
KeyError	If the service type is not available

Example

logger = ctx.get(Logger)
database = ctx.get(Database)

with_service(t, v)

Convenient alias for add() - add a service to the context.

Parameters:

Name	Type	Description	Default
t	type[A]	The type of the service	required
v	A	The service instance	required

Returns:

Type	Description
'Context'	New context containing the added service

Scope API

effectpy.scope

Scope

Resource lifecycle manager with guaranteed cleanup.

Scope ensures that resources are cleaned up in LIFO order (Last In, First Out) even when exceptions occur. This is essential for safe resource management in async applications.

Example

scope = Scope()

# Add resources to scope
db = Database("postgresql://...")
scope.add_finalizer(db.close)

cache = Cache("redis://...")
scope.add_finalizer(cache.close)

try:
    # Use resources...
    pass
finally:
    await scope.close()  # Closes cache, then db (LIFO order)

add_finalizer(fin) async

Add a cleanup function to be called when the scope closes.

Finalizers are called in LIFO order (reverse of addition order). If the scope is already closed, the finalizer is executed immediately.

Parameters:

Name	Type	Description	Default
fin	Callable[[], Awaitable[None]]	Async function to call during cleanup	required

Example

scope = Scope()

# Add cleanup functions
scope.add_finalizer(lambda: database.close())
scope.add_finalizer(lambda: cache.disconnect())

# Later...
await scope.close()  # Calls cache.disconnect(), then database.close()

close() async

Close the scope and run all finalizers in LIFO order.

Finalizers are executed in reverse order of addition. Even if some finalizers fail, all remaining finalizers will still be executed. The scope becomes closed and cannot accept new finalizers.

Example

scope = Scope()
scope.add_finalizer(cleanup1)
scope.add_finalizer(cleanup2)

await scope.close()  # Runs cleanup2, then cleanup1