definition

async_btree.definition

Common definition.

CallableFunction Type:

Specify something callable with or without async:

CallableFunction = Union[Callable[..., Awaitable[Any]], Callable]

Function signature of async function implementation:

AsyncInnerFunction = Callable[[], Awaitable[Any]]

Attributes

AsyncCallableFunction = Callable[..., Awaitable[Any]] module-attribute

Async callable.

AsyncInnerFunction = Callable[[], Awaitable[Any]] module-attribute

Function signature of async function implementation.

CallableFunction = Callable[..., Awaitable[Any]] | Callable module-attribute

Something callable with or without async.

FAILURE = not SUCCESS module-attribute

Failure constant.

SUCCESS = True module-attribute

Success constant.

Classes

ControlFlowException

Bases: Exception

Wraps an exception to give it falsy meaning without losing the original cause.

Instances are always falsy (bool(e) returns False), so they can be returned as a FAILURE status while still carrying the original exception.

Source code in async_btree/definition.py

class ControlFlowException(Exception):
    """Wraps an exception to give it falsy meaning without losing the original cause.

    Instances are always falsy (`bool(e)` returns `False`), so they can be
    returned as a FAILURE status while still carrying the original exception.
    """

    def __init__(self, exception: Exception):
        """Initialize with the original exception.

        Args:
            exception (Exception): the original exception to wrap.
        """
        super().__init__()

        self.exception = exception

    def __bool__(self):
        return False

    def __repr__(self):
        return self.exception.__repr__()

    def __str__(self):
        return self.exception.__str__()

    @classmethod
    def instantiate(cls, exception: Exception) -> ControlFlowException:
        """Return `exception` unchanged if already a `ControlFlowException`, otherwise wrap it.

        Args:
            exception (Exception): the exception to wrap if needed.

        Returns:
            (ControlFlowException): a falsy exception suitable for use as FAILURE.
        """
        return exception if isinstance(exception, ControlFlowException) else ControlFlowException(exception=exception)

Functions

__init__(exception)

Initialize with the original exception.

Parameters:

Name	Type	Description	Default
exception	Exception	the original exception to wrap.	required

Source code in async_btree/definition.py

def __init__(self, exception: Exception):
    """Initialize with the original exception.

    Args:
        exception (Exception): the original exception to wrap.
    """
    super().__init__()

    self.exception = exception

instantiate(exception) classmethod

Return exception unchanged if already a ControlFlowException, otherwise wrap it.

Parameters:

Name	Type	Description	Default
exception	Exception	the exception to wrap if needed.	required

Returns:

Type	Description
ControlFlowException	a falsy exception suitable for use as FAILURE.

Source code in async_btree/definition.py

@classmethod
def instantiate(cls, exception: Exception) -> ControlFlowException:
    """Return `exception` unchanged if already a `ControlFlowException`, otherwise wrap it.

    Args:
        exception (Exception): the exception to wrap if needed.

    Returns:
        (ControlFlowException): a falsy exception suitable for use as FAILURE.
    """
    return exception if isinstance(exception, ControlFlowException) else ControlFlowException(exception=exception)

NodeMetadata

Bases: NamedTuple

Metadata attached to a node function describing its name, properties, and child edges.

Used by analyze() and stringify_analyze() to build and display the abstract tree.

Attributes:

Name	Type	Description
name	str	display name of the node.
properties	List[str] | None	names of scalar attributes to include in the tree view.
edges	List[str] | None	names of child-bearing attributes. When None, analyze() falls back to ["child", "children", "_child", "_children"].

Source code in async_btree/definition.py

class NodeMetadata(NamedTuple):
    """Metadata attached to a node function describing its name, properties, and child edges.

    Used by `analyze()` and `stringify_analyze()` to build and display the abstract tree.

    Attributes:
        name (str): display name of the node.
        properties (List[str] | None): names of scalar attributes to include in the tree view.
        edges (List[str] | None): names of child-bearing attributes. When `None`, `analyze()`
            falls back to `["child", "children", "_child", "_children"]`.
    """

    name: str
    properties: list[str] | None = None
    edges: list[str] | None = None

    @classmethod
    def alias(cls, name: str, node: NodeMetadata, properties: list[str] | None = None) -> NodeMetadata:
        """Return a copy of `node` with a new name and optional property override.

        Args:
            name (str): new display name.
            node (NodeMetadata): source metadata to copy edges from.
            properties (List[str] | None): if given, replaces `node.properties`.

        Returns:
            (NodeMetadata): new instance with updated name and properties.
        """
        return NodeMetadata(
            name=name,
            properties=properties if properties else node.properties,
            edges=node.edges,
        )

Functions

alias(name, node, properties=None) classmethod

Return a copy of node with a new name and optional property override.

Parameters:

Name	Type	Description	Default
name	str	new display name.	required
node	NodeMetadata	source metadata to copy edges from.	required
properties	List[str] | None	if given, replaces node.properties.	None

Returns:

Type	Description
NodeMetadata	new instance with updated name and properties.

Source code in async_btree/definition.py

@classmethod
def alias(cls, name: str, node: NodeMetadata, properties: list[str] | None = None) -> NodeMetadata:
    """Return a copy of `node` with a new name and optional property override.

    Args:
        name (str): new display name.
        node (NodeMetadata): source metadata to copy edges from.
        properties (List[str] | None): if given, replaces `node.properties`.

    Returns:
        (NodeMetadata): new instance with updated name and properties.
    """
    return NodeMetadata(
        name=name,
        properties=properties if properties else node.properties,
        edges=node.edges,
    )

Functions

alias_node_metadata(target, name, properties=None)

Mutate target.__node_metadata in place to apply an alias name and optional properties.

Parameters:

Name	Type	Description	Default
target	CallableFunction	function whose __node_metadata will be updated.	required
name	str	new display name to assign.	required
properties	Optional[List[str]]	if given, replaces the existing properties list.	None

Returns:

Type	Description
CallableFunction	target with its __node_metadata updated.

Source code in async_btree/definition.py

def alias_node_metadata(target: CallableFunction, name: str, properties: list[str] | None = None) -> CallableFunction:
    """Mutate `target.__node_metadata` in place to apply an alias name and optional properties.

    Args:
        target (CallableFunction): function whose `__node_metadata` will be updated.
        name (str): new display name to assign.
        properties (Optional[List[str]]): if given, replaces the existing properties list.

    Returns:
        (CallableFunction): `target` with its `__node_metadata` updated.
    """
    dfunc = _attr_decorator(target)
    dfunc.__node_metadata = NodeMetadata.alias(name=name, node=dfunc.__node_metadata, properties=properties)
    return dfunc

get_function_name(target, default_name='anonymous')

Returns a function name.

Parameters:

Name	Type	Description	Default
target	CallableFunction	function to analyze.	required
default_name	str	default name 'anonymous'	'anonymous'

Returns:

Type	Description
str	function name

Source code in async_btree/definition.py

def get_function_name(target: Callable, default_name: str = "anonymous") -> str:
    """Returns a function name.

    Args:
        target (CallableFunction): function to analyze.
        default_name (str): default name 'anonymous'

    Returns:
        (str): function name

    """
    return getattr(target, "__name__", default_name).lstrip("_")

get_node_metadata(target)

Return the NodeMetadata instance attached to target.

Parameters:

Name	Type	Description	Default
target	CallableFunction	function decorated with @node_metadata.	required

Returns:

Type	Description
NodeMetadata	the metadata attached to target.

Raises:

Type	Description
RuntimeError	if target has no __node_metadata attribute, or if it is not a NodeMetadata instance.

Source code in async_btree/definition.py

def get_node_metadata(target: CallableFunction) -> NodeMetadata:
    """Return the `NodeMetadata` instance attached to `target`.

    Args:
        target (CallableFunction): function decorated with `@node_metadata`.

    Returns:
        (NodeMetadata): the metadata attached to `target`.

    Raises:
        RuntimeError: if `target` has no `__node_metadata` attribute, or if it is
            not a `NodeMetadata` instance.
    """
    node = getattr(target, "__node_metadata", False)
    if not isinstance(node, NodeMetadata):
        raise RuntimeError(f"attr __node_metadata of {target} is not a NodeMetadata!")
    return cast("NodeMetadata", node)

node_metadata(name=None, properties=None, edges=None)

Decorator that attaches NodeMetadata to a function as __node_metadata.

Parameters:

Name	Type	Description	Default
name	Optional[str]	override display name; defaults to the function name left-stripped of leading underscores.	None
properties	Optional[List[str]]	names of scalar attributes to expose in the tree view.	None
edges	Optional[List[str]]	names of child-bearing attributes. When None, analyze() falls back to ["child", "children", "_child", "_children"].	None

Returns:

Type	Description
Callable[[Callable[P, R]], FunctionWithMetadata[P, R]]	the decorator function.

Source code in async_btree/definition.py

def node_metadata(
    name: str | None = None,
    properties: list[str] | None = None,
    edges: list[str] | None = None,
) -> Callable[[Callable[P, R]], FunctionWithMetadata[P, R]]:
    """Decorator that attaches `NodeMetadata` to a function as `__node_metadata`.

    Args:
        name (Optional[str]): override display name; defaults to the function name
            left-stripped of leading underscores.
        properties (Optional[List[str]]): names of scalar attributes to expose in the tree view.
        edges (Optional[List[str]]): names of child-bearing attributes. When `None`, `analyze()`
            falls back to `["child", "children", "_child", "_children"]`.

    Returns:
        the decorator function.
    """

    def decorate_function(function: Callable[P, R]) -> FunctionWithMetadata[P, R]:
        dfunc = _attr_decorator(function)

        dfunc.__node_metadata = getattr(
            dfunc,
            "__node_metadata",
            NodeMetadata(
                name=name if name else get_function_name(target=dfunc),
                properties=properties,
                edges=edges,
            ),
        )
        return cast("FunctionWithMetadata[P, R]", dfunc)

    return decorate_function