Framework context manager

FrameworkContextManager

A manager class to set, get and clear thread level context.

Source code in agentuniverse/base/context/framework_context_manager.py

Python

@singleton
class FrameworkContextManager:
    """A manager class to set, get and clear thread level context."""

    def __init__(self):
        """Init an empty context variable dict and a thread lock used when
        add new key to this dict."""
        self.__context_dict: Dict[str, ContextVar] = {}
        self.__dict_edit_lock = threading.Lock()

    def is_context_exist(self, var_name: str) -> bool:
        """Judge whether context variable exist in current context.

        Args:
            var_name (`str`):
                Name of the context variable.

        Returns:
            A boolean value indicating whether the variable exists.
        """
        return var_name in self.__context_dict

    def set_context(self, var_name: str, var_value: Any) -> Token:
        """Set a context variable value.

        Args:
            var_name (`str`):
                Name of the context variable.
            var_value (`Any`):
                Value of the context variable.
        """
        if var_name not in self.__context_dict:
            with self.__dict_edit_lock:
                if var_name not in self.__context_dict:
                    self.__context_dict[var_name] = ContextVar(var_name)
        return self.__context_dict[var_name].set(var_value)

    def get_context(self,
                    var_name: str,
                    default_value: Any = None):
        """Get a context variable value.

        Args:
            var_name (`str`):
                Name of the context variable.
            default_value (`Any`, defaults to `None`):
                Value to be returned if target context variable doesn't exist.
        """
        if var_name not in self.__context_dict:
            return default_value
        return self.__context_dict[var_name].get(default_value)

    def del_context(self, var_name: str):
        """Set a context variable to None.

        Args:
            var_name (`str`): Name of the context variable.
        """
        if self.is_context_exist(var_name):
            self.__context_dict[var_name].set(None)

    def reset_context(self, var_name: str, token: Token):
        """Reset a context variable using given token.

        Args:
            var_name (`str`):
                Name of the context variable.
            token (`Token`):
                Token used to reset a context variable.
        """
        self.__context_dict[var_name].reset(token)

__init__()

Init an empty context variable dict and a thread lock used when add new key to this dict.

Source code in agentuniverse/base/context/framework_context_manager.py

Python

def __init__(self):
    """Init an empty context variable dict and a thread lock used when
    add new key to this dict."""
    self.__context_dict: Dict[str, ContextVar] = {}
    self.__dict_edit_lock = threading.Lock()

del_context(var_name)

Set a context variable to None.

Parameters:

Name	Type	Description	Default
var_name	`str`	Name of the context variable.	required

Source code in agentuniverse/base/context/framework_context_manager.py

Python

def del_context(self, var_name: str):
    """Set a context variable to None.

    Args:
        var_name (`str`): Name of the context variable.
    """
    if self.is_context_exist(var_name):
        self.__context_dict[var_name].set(None)

get_context(var_name, default_value=None)

Get a context variable value.

Parameters:

Name	Type	Description	Default
var_name	`str`	Name of the context variable.	required
default_value	`Any`, defaults to `None`	Value to be returned if target context variable doesn't exist.	None

Source code in agentuniverse/base/context/framework_context_manager.py

Python

def get_context(self,
                var_name: str,
                default_value: Any = None):
    """Get a context variable value.

    Args:
        var_name (`str`):
            Name of the context variable.
        default_value (`Any`, defaults to `None`):
            Value to be returned if target context variable doesn't exist.
    """
    if var_name not in self.__context_dict:
        return default_value
    return self.__context_dict[var_name].get(default_value)

is_context_exist(var_name)

Judge whether context variable exist in current context.

Parameters:

Name	Type	Description	Default
var_name	`str`	Name of the context variable.	required

Returns:

Type	Description
bool	A boolean value indicating whether the variable exists.

Source code in agentuniverse/base/context/framework_context_manager.py

Python

def is_context_exist(self, var_name: str) -> bool:
    """Judge whether context variable exist in current context.

    Args:
        var_name (`str`):
            Name of the context variable.

    Returns:
        A boolean value indicating whether the variable exists.
    """
    return var_name in self.__context_dict

reset_context(var_name, token)

Reset a context variable using given token.

Parameters:

Name	Type	Description	Default
var_name	`str`	Name of the context variable.	required
token	`Token`	Token used to reset a context variable.	required

Source code in agentuniverse/base/context/framework_context_manager.py

Python

def reset_context(self, var_name: str, token: Token):
    """Reset a context variable using given token.

    Args:
        var_name (`str`):
            Name of the context variable.
        token (`Token`):
            Token used to reset a context variable.
    """
    self.__context_dict[var_name].reset(token)

set_context(var_name, var_value)

Set a context variable value.

Parameters:

Name	Type	Description	Default
var_name	`str`	Name of the context variable.	required
var_value	`Any`	Value of the context variable.	required

Source code in agentuniverse/base/context/framework_context_manager.py

Python

def set_context(self, var_name: str, var_value: Any) -> Token:
    """Set a context variable value.

    Args:
        var_name (`str`):
            Name of the context variable.
        var_value (`Any`):
            Value of the context variable.
    """
    if var_name not in self.__context_dict:
        with self.__dict_edit_lock:
            if var_name not in self.__context_dict:
                self.__context_dict[var_name] = ContextVar(var_name)
    return self.__context_dict[var_name].set(var_value)