Skip to content

ContextRepo

faststream.utils.ContextRepo #

ContextRepo()

Bases: Singleton

A class to represent a context repository.

METHOD DESCRIPTION
__init__

initializes the ContextRepo object

set_global

sets a global context variable

reset_global

resets a global context variable

set_local

sets a local context variable

reset_local

resets a local context variable

get_local

gets the value of a local context variable

clear

clears the global and scope context

get

gets the value of a context variable

__getattr__

gets the value of a context variable using attribute access

context

gets the current context as a dictionary

scope

creates a context scope for a specific key and value

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Initialize the class.

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/utils/context/main.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
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    self._global_context = {"context": self}
    self._scope_context = {}

context property #

context: AnyDict

clear #

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

get #

get(key: str, default: Any = None) -> Any

Get the value associated with a key.

PARAMETER DESCRIPTION
key

The key to retrieve the value for.

TYPE: str

RETURNS DESCRIPTION
Any

The value associated with the key.

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/utils/context/main.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.

    Returns:
        The value associated with the key.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    return self._global_context.get(key, self.get_local(key, default))

get_local #

get_local(key: str, default: Any = None) -> Any

Get the value of a local variable.

PARAMETER DESCRIPTION
key

The key of the local variable to retrieve.

TYPE: str

RETURNS DESCRIPTION
Any

The value of the local variable.

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/utils/context/main.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.

    Returns:
        The value of the local variable.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    context_var = self._scope_context.get(key)
    if context_var is not None:  # pragma: no branch
        return context_var.get()
    else:
        return default

reset_global #

reset_global(key: str) -> None

Resets a key in the global context.

PARAMETER DESCRIPTION
key

The key to reset in the global context.

TYPE: str

RETURNS DESCRIPTION
None

None

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/utils/context/main.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
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    self._global_context.pop(key, None)

reset_local #

reset_local(key: str, tag: Token[Any]) -> None

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

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/utils/context/main.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
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    self._scope_context[key].reset(tag)

scope #

scope(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.

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

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/utils/context/main.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
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    token = self.set_local(key, value)
    try:
        yield
    finally:
        self.reset_local(key, token)

set_global #

set_global(key: str, v: Any) -> None

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.

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/utils/context/main.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.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    self._global_context[key] = v

set_local #

set_local(key: str, value: T) -> Token[T]

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[T]

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

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/utils/context/main.py
def set_local(self, key: str, value: T) -> "Token[T]":
    """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.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    context_var = self._scope_context.get(key)
    if context_var is None:
        context_var = ContextVar(key, default=None)
        self._scope_context[key] = context_var
    return context_var.set(value)

Last update: 2023-11-13