Skip to content

ContextRepo

faststream.utils.context.ContextRepo #

ContextRepo()

Bases: Singleton

A class to represent a context repository.

Initialize the class.

ATTRIBUTE DESCRIPTION
_global_context

a dictionary representing the global context

_scope_context

a dictionary representing the scope context

Source code in faststream/utils/context/repository.py
def __init__(self) -> None:
    """Initialize the class.

    Attributes:
        _global_context : a dictionary representing the global context
        _scope_context : a dictionary representing the scope context
    """
    self._global_context = {"context": self}
    self._scope_context = {}

context property #

context

set_global #

set_global(key, v)

Sets a value in the global context.

PARAMETER DESCRIPTION
key

The key to set in the global context.

TYPE: str

v

The value to set.

TYPE: Any

RETURNS DESCRIPTION
None

None.

Source code in faststream/utils/context/repository.py
def set_global(self, key: str, v: Any) -> None:
    """Sets a value in the global context.

    Args:
        key: The key to set in the global context.
        v: The value to set.

    Returns:
        None.
    """
    self._global_context[key] = v

reset_global #

reset_global(key)

Resets a key in the global context.

PARAMETER DESCRIPTION
key

The key to reset in the global context.

TYPE: str

RETURNS DESCRIPTION
None

None

Source code in faststream/utils/context/repository.py
def reset_global(self, key: str) -> None:
    """Resets a key in the global context.

    Args:
        key (str): The key to reset in the global context.

    Returns:
        None
    """
    self._global_context.pop(key, None)

set_local #

set_local(key, value)

Set a local context variable.

PARAMETER DESCRIPTION
key

The key for the context variable.

TYPE: str

value

The value to set for the context variable.

TYPE: T

RETURNS DESCRIPTION
Token[Any]

Token[T]: A token representing the context variable.

Source code in faststream/utils/context/repository.py
def set_local(self, key: str, value: Any) -> "Token[Any]":
    """Set a local context variable.

    Args:
        key (str): The key for the context variable.
        value (T): The value to set for the context variable.

    Returns:
        Token[T]: A token representing the context variable.
    """
    context_var = self._scope_context.get(key)
    if context_var is None:
        context_var = ContextVar(key, default=EMPTY)
        self._scope_context[key] = context_var
    return context_var.set(value)

reset_local #

reset_local(key, tag)

Resets the local context for a given key.

PARAMETER DESCRIPTION
key

The key to reset the local context for.

TYPE: str

tag

The tag associated with the local context.

TYPE: Token[Any]

RETURNS DESCRIPTION
None

None

Source code in faststream/utils/context/repository.py
def reset_local(self, key: str, tag: "Token[Any]") -> None:
    """Resets the local context for a given key.

    Args:
        key (str): The key to reset the local context for.
        tag (Token[Any]): The tag associated with the local context.

    Returns:
        None
    """
    self._scope_context[key].reset(tag)

get_local #

get_local(key, default=None)

Get the value of a local variable.

PARAMETER DESCRIPTION
key

The key of the local variable to retrieve.

TYPE: str

default

The default value to return if the local variable is not found.

TYPE: Any DEFAULT: None

RETURNS DESCRIPTION
Any

The value of the local variable.

Source code in faststream/utils/context/repository.py
def get_local(self, key: str, default: Any = None) -> Any:
    """Get the value of a local variable.

    Args:
        key: The key of the local variable to retrieve.
        default: The default value to return if the local variable is not found.

    Returns:
        The value of the local variable.
    """
    value = default
    if (context_var := self._scope_context.get(key)) is not None and (
        context_value := context_var.get()
    ) is not EMPTY:
        value = context_value
    return value

scope #

scope(key, value)

Sets a local variable and yields control to the caller. After the caller is done, the local variable is reset.

PARAMETER DESCRIPTION
key

The key of the local variable

TYPE: str

value

The value to set the local variable to

TYPE: Any

YIELDS DESCRIPTION
Iterator[None]

None

RETURNS DESCRIPTION
Iterator[None]

An iterator that yields None

Source code in faststream/utils/context/repository.py
@contextmanager
def scope(self, key: str, value: Any) -> Iterator[None]:
    """Sets a local variable and yields control to the caller. After the caller is done, the local variable is reset.

    Args:
        key: The key of the local variable
        value: The value to set the local variable to

    Yields:
        None

    Returns:
        An iterator that yields None
    """
    token = self.set_local(key, value)
    try:
        yield
    finally:
        self.reset_local(key, token)

get #

get(key, default=None)

Get the value associated with a key.

PARAMETER DESCRIPTION
key

The key to retrieve the value for.

TYPE: str

default

The default value to return if the key is not found.

TYPE: Any DEFAULT: None

RETURNS DESCRIPTION
Any

The value associated with the key.

Source code in faststream/utils/context/repository.py
def get(self, key: str, default: Any = None) -> Any:
    """Get the value associated with a key.

    Args:
        key: The key to retrieve the value for.
        default: The default value to return if the key is not found.

    Returns:
        The value associated with the key.
    """
    if (glob := self._global_context.get(key, EMPTY)) is EMPTY:
        return self.get_local(key, default)
    else:
        return glob

resolve #

resolve(argument)

Resolve the context of an argument.

PARAMETER DESCRIPTION
argument

A string representing the argument.

TYPE: str

RETURNS DESCRIPTION
Any

The resolved context of the argument.

RAISES DESCRIPTION
(AttributeError, KeyError)

If the argument does not exist in the context.

Source code in faststream/utils/context/repository.py
def resolve(self, argument: str) -> Any:
    """Resolve the context of an argument.

    Args:
        argument: A string representing the argument.

    Returns:
        The resolved context of the argument.

    Raises:
        AttributeError, KeyError: If the argument does not exist in the context.
    """
    first, *keys = argument.split(".")

    if (v := self.get(first, EMPTY)) is EMPTY:
        raise KeyError(f"`{self.context}` does not contains `{first}` key")

    for i in keys:
        v = v[i] if isinstance(v, Mapping) else getattr(v, i)

    return v

clear #

clear()
Source code in faststream/utils/context/repository.py
def clear(self) -> None:
    self._global_context = {"context": self}
    self._scope_context.clear()