runner

async_btree.runner

Classes

BTreeRunner

Context manager that runs behavior trees against a configurable async backend.

On __enter__, a snapshot of the caller's ContextVar state is captured via copy_context(). Each run() call executes in an isolated copy of that snapshot — mutations inside a run do not escape to the caller and do not accumulate across successive run() calls.

Parameters:

Name	Type	Description	Default
backend	Backend	async runtime to use — "asyncio" (default), "trio", or "asyncio+uvloop".	'asyncio'

Source code in async_btree/runner.py

class BTreeRunner:
    """Context manager that runs behavior trees against a configurable async backend.

    On `__enter__`, a snapshot of the caller's `ContextVar` state is captured via
    `copy_context()`. Each `run()` call executes in an isolated copy of that snapshot —
    mutations inside a run do not escape to the caller and do not accumulate across
    successive `run()` calls.

    Args:
        backend (Backend): async runtime to use — "asyncio" (default), "trio", or
            "asyncio+uvloop".
    """

    def __init__(self, backend: Backend = "asyncio") -> None:
        self._anyio_backend, self._anyio_backend_options = _BACKEND_MAP[backend]
        self._context: contextvars.Context | None = None

    def __enter__(self) -> BTreeRunner:
        self._context = copy_context()
        return self

    def __exit__(self, *_: Any) -> None:
        self._context = None

    def run(self, target: Callable[..., Awaitable[R]], *args: Any, **kwargs: Any) -> R:
        """Run an async callable to completion using the configured backend.

        Must be called within a `with BTreeRunner() as runner:` block.

        Each call runs in an isolated copy of the context captured at `__enter__`.
        ContextVar mutations inside `target` do not escape to the caller and do not
        accumulate across successive `run()` calls.

        Args:
            target: async callable (coroutine function)
            *args: positional arguments passed to target
            **kwargs: keyword arguments passed to target

        Returns:
            whatever target returns

        Raises:
            RuntimeError: if called outside of the context manager
        """
        if self._context is None:
            raise RuntimeError("BTreeRunner.run() must be called within a 'with' block")
        return self._context.run(
            anyio.run, target, *args, backend=self._anyio_backend, backend_options=self._anyio_backend_options, **kwargs
        )

Functions

run(target, *args, **kwargs)

Run an async callable to completion using the configured backend.

Must be called within a with BTreeRunner() as runner: block.

Each call runs in an isolated copy of the context captured at __enter__. ContextVar mutations inside target do not escape to the caller and do not accumulate across successive run() calls.

Parameters:

Name	Type	Description	Default
target	Callable[..., Awaitable[R]]	async callable (coroutine function)	required
*args	Any	positional arguments passed to target	()
**kwargs	Any	keyword arguments passed to target	{}

Returns:

Type	Description
R	whatever target returns

Raises:

Type	Description
RuntimeError	if called outside of the context manager

Source code in async_btree/runner.py

def run(self, target: Callable[..., Awaitable[R]], *args: Any, **kwargs: Any) -> R:
    """Run an async callable to completion using the configured backend.

    Must be called within a `with BTreeRunner() as runner:` block.

    Each call runs in an isolated copy of the context captured at `__enter__`.
    ContextVar mutations inside `target` do not escape to the caller and do not
    accumulate across successive `run()` calls.

    Args:
        target: async callable (coroutine function)
        *args: positional arguments passed to target
        **kwargs: keyword arguments passed to target

    Returns:
        whatever target returns

    Raises:
        RuntimeError: if called outside of the context manager
    """
    if self._context is None:
        raise RuntimeError("BTreeRunner.run() must be called within a 'with' block")
    return self._context.run(
        anyio.run, target, *args, backend=self._anyio_backend, backend_options=self._anyio_backend_options, **kwargs
    )