`asphalt.core.context`

class asphalt.core.context.Context(parent=None)

Contexts give request handlers and callbacks access to resources.

Contexts are stacked in a way that accessing an attribute that is not present in the current context causes the attribute to be looked up in the parent instance and so on, until the attribute is found (or AttributeError is raised).

Parameters:

parent (Optional[Context]) – the parent context, if any

Variables:

parent (Context) – the parent context, if any
resource_added (Signal) – a signal (ResourceEvent) dispatched when a resource has been published in this context

add_resource(value, name='default', context_attr=None, types=())

Add a resource to this context.

This will cause a resource_added event to be dispatched.

Parameters:

value – the actual resource value
name (str) – name of this resource (unique among all its registered types within a single context)
context_attr (str | None) – (deprecated) name of the context attribute this resource will be accessible as
types (Union[type, Sequence[type]]) – type(s) to register the resource as (omit to use the type of value)

Raises:

asphalt.core.context.ResourceConflict – if the resource conflicts with an existing one in any way

Return type:

None

add_resource_factory(factory_callback, types=None, name='default', context_attr=None)

Add a resource factory to this context.

This will cause a resource_added event to be dispatched.

A resource factory is a callable that generates a “contextual” resource when it is requested by either using any of the methods get_resource(), require_resource() or request_resource() or its context attribute is accessed.

The type(s) of the generated resources need to be specified, either by passing the types argument, or by adding a return type annotation to the factory function. If the generated resource needs to be registered as multiple types, you can use Union (e.g. Union[str, int]) or a union type (e.g. str | int; requires from __future__ import annotations on earlier than Python 3.10).

When a new resource is created in this manner, it is always bound to the context through it was requested, regardless of where in the chain the factory itself was added to.

Parameters:

factory_callback (factory_callback_type) – a (non-coroutine) callable that takes a context instance as argument and returns the created resource object
types (Union[type, Sequence[Type], None]) – one or more types to register the generated resource as on the target context (can be omitted if the factory callable has a return type annotation)
name (str) – name of the resource that will be created in the target context
context_attr (str | None) – name of the context attribute the created resource will be accessible as

Raises:

asphalt.core.context.ResourceConflict – if there is an existing resource factory for the given type/name combinations or the given context variable

Return type:

None

add_teardown_callback(callback, pass_exception=False)

Add a callback to be called when this context closes.

This is intended for cleanup of resources, and the list of callbacks is processed in the reverse order in which they were added, so the last added callback will be called first.

The callback may return an awaitable. If it does, the awaitable is awaited on before calling any further callbacks.

Parameters:

callback (Callable) – a callable that is called with either no arguments or with the exception that ended this context, based on the value of pass_exception
pass_exception (bool) – True to pass the callback the exception that ended this context (or None if the context ended cleanly)

Return type:

None

call_async(func, *args, **kwargs)

Call the given callable in the event loop thread.

This method lets you call asynchronous code from a worker thread. Do not use it from within the event loop thread.

If the callable returns an awaitable, it is resolved before returning to the caller.

Parameters:

func (Callable[..., TypeVar(T_Retval)]) – a regular function or a coroutine function
args – positional arguments to call the callable with
kwargs – keyword arguments to call the callable with

Return type:

TypeVar(T_Retval)

Returns:

the return value of the call

call_in_executor(func, *args, executor=None, **kwargs)

Call the given callable in an executor.

Parameters:

func (Callable[..., TypeVar(T_Retval)]) – the callable to call
args – positional arguments to call the callable with
executor (Union[Executor, str, None]) – either an Executor instance, the resource name of one or None to use the event loop’s default executor
kwargs – keyword arguments to call the callable with

Return type:

Awaitable[TypeVar(T_Retval)]

Returns:

an awaitable that resolves to the return value of the call

async close(exception=None)

Close this context and call any necessary resource teardown callbacks.

If a teardown callback returns an awaitable, the return value is awaited on before calling any further teardown callbacks.

All callbacks will be processed, even if some of them raise exceptions. If at least one callback raised an error, this method will raise a TeardownError at the end.

After this method has been called, resources can no longer be requested or published on this context.

Parameters:: exception (BaseException | None) – the exception, if any, that caused this context to be closed
Raises:: .TeardownError – if one or more teardown callbacks raise an exception
Return type:: None

property closed: bool: Return True if the teardown process has at least been initiated, False otherwise.

property context_chain: List[Context]: Return a list of contexts starting from this one, its parent and so on.

get_resource(type, name='default')

Look up a resource in the chain of contexts.

Parameters:

type (Type[TypeVar(T_Resource)]) – type of the requested resource
name (str) – name of the requested resource

Return type:

Optional[TypeVar(T_Resource)]

Returns:

the requested resource, or None if none was available

get_resources(type)

Retrieve all the resources of the given type in this context and its parents.

Any matching resource factories are also triggered if necessary.

Parameters:: type (Type[TypeVar(T_Resource)]) – type of the resources to get
Return type:: Set[TypeVar(T_Resource)]
Returns:: a set of all found resources of the given type

property loop: AbstractEventLoop: Return the event loop associated with this context.

property parent: Context | None: Return the parent context, or None if there is no parent.

async request_resource(type, name='default')

Look up a resource in the chain of contexts.

This is like get_resource() except that if the resource is not already available, it will wait for one to become available.

Parameters:

type (Type[TypeVar(T_Resource)]) – type of the requested resource
name (str) – name of the requested resource

Return type:

TypeVar(T_Resource)

Returns:

the requested resource

require_resource(type, name='default')

Look up a resource in the chain of contexts and raise an exception if it is not found.

This is like get_resource() except that instead of returning None when a resource is not found, it will raise ResourceNotFound.

Parameters:

type (Type[TypeVar(T_Resource)]) – type of the requested resource
name (str) – name of the requested resource

Return type:

TypeVar(T_Resource)

Returns:

the requested resource

Raises:

asphalt.core.context.ResourceNotFound – if a resource of the given type and name was not found

threadpool(executor=None)

Return an asynchronous context manager that runs the block in a (thread pool) executor.

Parameters:: executor (Union[Executor, str, None]) – either an Executor instance, the resource name of one or None to use the event loop’s default executor
Returns:: an asynchronous context manager

exception asphalt.core.context.NoCurrentContext: Raised by :func: current_context when there is no active context.

exception asphalt.core.context.ResourceConflict: Raised when a new resource that is being published conflicts with an existing resource or context variable.

class asphalt.core.context.ResourceEvent(source, topic, types, name, is_factory)

Dispatched when a resource or resource factory has been added to a context.

Variables:

resource_types (Tuple[type, ...]) – types the resource was registered under
name (str) – name of the resource
is_factory (bool) – True if a resource factory was added, False if a regular resource was added

exception asphalt.core.context.ResourceNotFound(type, name): Raised when a resource request cannot be fulfilled within the allotted time.

exception asphalt.core.context.TeardownError(exceptions)

Raised after context teardown when one or more teardown callbacks raised an exception.

Variables:: exceptions (List[Exception]) – exceptions raised during context teardown, in the order in which they were raised

asphalt.core.context.context_teardown(func)

Wrap an async generator function to execute the rest of the function at context teardown.

This function returns an async function, which, when called, starts the wrapped async generator. The wrapped async function is run until the first yield statement When the context is being torn down, the exception that ended the context, if any, is sent to the generator.

For example:

class SomeComponent(Component):
    @context_teardown
    async def start(self, ctx: Context):
        service = SomeService()
        ctx.add_resource(service)
        exception = yield
        service.stop()

Parameters:: func (Callable[[T_Context], AsyncGenerator[None, Exception | None]] | Callable[[T_Self, T_Context], AsyncGenerator[None, Exception | None]]) – an async generator function
Return type:: Callable[[T_Context], Coroutine[Any, Any, None]] | Callable[[T_Self, T_Context], Coroutine[Any, Any, None]]
Returns:: an async function

asphalt.core.context.current_context()

Return the currently active context.

Raises:: NoCurrentContext – if there is no active context
Return type:: Context

asphalt.core.context.executor(arg=None)

Decorate a function so that it runs in an Executor.

If a resource name is given, the first argument must be a Context.

Usage:

@executor
def should_run_in_executor():
    ...

With a resource name:

@executor('resourcename')
def should_run_in_executor(ctx):
    ...

Parameters:: arg (Union[Executor, str, None, Callable[..., TypeVar(T_Retval)]]) – a callable to decorate, an Executor instance, the resource name of one or None to use the event loop’s default executor
Returns:: the wrapped function

asphalt.core.context.get_resource(type, name='default')

Shortcut for current_context().get_resource(...).

Return type:: T_Resource | None

asphalt.core.context.get_resources(type): Shortcut for current_context().get_resources(...).

asphalt.core.context.inject(func)

Wrap the given coroutine function for use with dependency injection.

Parameters with dependencies need to be annotated and have resource() as the default value. When the wrapped function is called, values for such parameters will be automatically filled in by calling require_resource() using the parameter’s type annotation and the resource name passed to resource() (or "default") as the arguments.

Any forward references among the type annotations are resolved on the first call to the wrapped function.

Return type:: Callable[[ParamSpec(P)], Any]

asphalt.core.context.require_resource(type, name='default')

Shortcut for current_context().require_resource(...).

Return type:: TypeVar(T_Resource)

asphalt.core.context.resource(name='default')

Marker for declaring a parameter for dependency injection via inject().

Parameters:: name (str) – the resource name (defaults to default)
Return type:: Any

asphalt.core.context

`asphalt.core.context`