Context Informants

Context informants tell the ContextManager about the context.

Each informant provides a unique string for each individual context. The context manager then caches objects by context. The context should be destroyed when it is no longer needed.

class autoinject.informants.ContextVarInformant[source]

Context informant for contextvars library.

destroy_self(context: ContextProtocol | Context | None = None)[source]

Destroy the context related to the contextvars context passed or the current one if None.

get_cache_id() str[source]

Obtain the current context ID from the contextvars.

class autoinject.informants.ContextVarManager(contextvar_informant: ContextVarInformant, context: Literal['_default', 'same', 'empty', 'copy'] | ContextProtocol | Context | None = '_default', suppress_exit_warning: bool = False)[source]

Wrapper around contexts to help manage issues with cleaning up dependencies.

copy(same_autoinject_context: bool = False)[source]

Make a copy of the context, with optional parameter to keep or reset the autoinjection variables.

static ensure_context_id(context: ContextProtocol | Context | None = None)[source]

Ensure there is a context ID.

static freshen_context(context: ContextProtocol | Context | None = None) Token[source]

Refresh the context by resetting the context ID.

static get_context_id(context: ContextProtocol | Context | None = None) str | None[source]

Retrieve the current context ID, but don’t set one if there isn’t one.

reset(var: ContextVar, token: Any)[source]

Reset a variable.

static restore_context_id(token, context: ContextProtocol | Context | None = None)[source]

Refresh the context by resetting the context ID.

run(cmd: Callable[[...], Q], *args, **kwargs) Q[source]

Run, in context if appropriate.

set(var: ContextVar, value: Any)[source]

Set a variable and return a token

class autoinject.informants.NamedSituationInformant(name: str = 'named_situation')[source]

A toy class for demonstrating how contexts work. The context can be changed using switch_context():

from autoinject import injector, NamedContextInformant

informant = NamedContextInformant()
injector.register_informant(informant)
# We are now in the "_default" context
informant.switch_context("alpha")
# We are now in the "alpha" context

In the above example, any calls to inject classes registered with a cache strategy of CONTEXT_CACHE (the default) will result in different obtains being returned for the _default context and the alpha context. If the class is registered with a strategy of GLOBAL_CACHE, the same object will be returned.

destroy_self()[source]

Destroy the current context

get_cache_id() str[source]

Provide the context ID to the ContextManager

switch_context(cache_id: str)[source]

Change the context ID

class autoinject.informants.SituationInformant(name: str | None = None)[source]

Base class for context informants

Parameters:

name (str) – A unique name for this informant. It will be used to assemble multiple contexts together.

check_expired_caches()[source]

Trigger to check for expired contexts so they can be cleaned-up from memory

destroy(cache_id: str)[source]

Remove all objects cached under the given context.

Parameters:

cache_id (str) – A value that would have been provided by get_context_id() to the ContextManager

abstractmethod get_cache_id() str[source]
Obtains a unique identifier for the current context. This is paired with the informant name to create a

unique string for each context.

Returns:

A unique string per context

Return type:

str

set_cache_manager(cache_manager: autoinject.cache_manager.CacheManager)[source]

Set the context manager. This is called by the ContextManager when the informant is registered.

Parameters:

cache_manager (autoinject.cache_manager.CacheManager) – The context manager

class autoinject.informants.ThreadedContextInformant[source]

Context informant for threading library

check_expired_caches()[source]

Since threads don’t reliably have a callback when they complete, we instead regularly monitor the active thread list and remove them as they complete to cut down on memory usage.

destroy_self(thread: Thread | None = None)[source]

Destroy the current thread context.

get_cache_id()[source]

Provide the context ID to the ContextManager