Skip to content

Index

async_btree

Declare async btree api.

Attributes

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

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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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
    )

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 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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 
67
68
69
70
71
72
73
74
75
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 
86
87
88
89
90
91
92
93
94
95
96
@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)

Node

Bases: NamedTuple

Resolved snapshot of a behaviour tree node, produced by analyze().

Holds the display name, resolved property values, and resolved child edges for a single node in the abstract tree.

Attributes:

Name Type Description
name str

display name of the node.
properties list[tuple[str, Any]]

resolved (name, value) pairs for scalar attributes.
edges list[tuple[str, list[Any]]]

resolved (edge_name, [Node, ...]) pairs for child relationships. Typed as list[Any] due to mypy #731 — actual element type is list[Node].
Source code in async_btree/analyze.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class Node(NamedTuple):
    """Resolved snapshot of a behaviour tree node, produced by `analyze()`.

    Holds the display name, resolved property values, and resolved child edges
    for a single node in the abstract tree.

    Attributes:
        name (str): display name of the node.
        properties (list[tuple[str, Any]]): resolved `(name, value)` pairs for scalar attributes.
        edges (list[tuple[str, list[Any]]]): resolved `(edge_name, [Node, ...])` pairs for
            child relationships. Typed as `list[Any]` due to
            [mypy #731](https://github.com/python/mypy/issues/731) — actual element type is
            `list[Node]`.
    """

    name: str
    properties: list[tuple[str, Any]]
    # edges: list[tuple[str, list['Node']]]
    # https://github.com/python/mypy/issues/731
    edges: list[tuple[str, list[Any]]]

    def __str__(self) -> str:
        """Return the stringified tree representation of this node."""
        return stringify_analyze(target=self)
Functions
__str__()

Return the stringified tree representation of this node.

Source code in async_btree/analyze.py 
34
35
36
def __str__(self) -> str:
    """Return the stringified tree representation of this node."""
    return stringify_analyze(target=self)

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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@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

action(target, **kwargs)

Declare an action leaf node.

Wraps target as an awaitable closure. Any exception raised by target is caught and re-raised as a ControlFlowException, giving it falsy meaning.

Parameters:

Name Type Description Default
target CallableFunction

sync or async callable to invoke.

 required
kwargs

keyword arguments forwarded to target on each call.

 {}

Returns:

Type Description
AsyncInnerFunction

an awaitable function.

Raises:

Type Description
ControlFlowException

wrapping any exception raised by target.
Source code in async_btree/leaf.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def action(target: CallableFunction, **kwargs) -> AsyncInnerFunction:
    """Declare an action leaf node.

    Wraps `target` as an awaitable closure. Any exception raised by `target`
    is caught and re-raised as a `ControlFlowException`, giving it falsy meaning.

    Args:
        target (CallableFunction): sync or async callable to invoke.
        kwargs: keyword arguments forwarded to `target` on each call.

    Returns:
        (AsyncInnerFunction): an awaitable function.

    Raises:
        ControlFlowException: wrapping any exception raised by `target`.
    """

    _target = to_async(target)

    @node_metadata(properties=["_target"])
    async def _action():
        try:
            return await _target(**kwargs)
        except Exception as e:
            raise ControlFlowException.instantiate(e) from e

    return _action

afilter(corofunc, iterable) async

Filter an iterable or an async iterable with an async function.

This simplify writing of filtering by a function on something iterable between 'async for ...' and 'for...' .

Parameters:

Name Type Description Default
corofunc Callable[[Any], Awaitable[bool]]

filter async function

 required
iterable Union[AsyncIterable, Iterable]

iterable or async iterable collection which will be applied.

 required

Returns:

Type Description
AsyncGenerator[Any]

an async iterator of item which satisfy corofunc(item) == True
Example

[i async for i in amap(inc, afilter(even, [0, 1, 2, 3, 4]))]

Source code in async_btree/utils.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
async def afilter(
    corofunc: Callable[[Any], Awaitable[bool]], iterable: AsyncIterable | Iterable
) -> AsyncGenerator[Any, None]:
    """Filter an iterable or an async iterable with an async function.

    This simplify writing of filtering by a function on something iterable
    between 'async for ...' and 'for...' .

    Args:
        corofunc (Callable[[Any], Awaitable[bool]]): filter async function
        iterable (Union[AsyncIterable, Iterable]): iterable or async iterable collection
            which will be applied.

    Returns:
        (AsyncGenerator[Any]): an async iterator of item which satisfy corofunc(item) == True

    Example:
        ```[i async for i in amap(inc, afilter(even, [0, 1, 2, 3, 4]))]```

    """
    if isinstance(iterable, AsyncIterable):
        async for item in iterable:
            if await corofunc(item):
                yield item
    else:
        for item in iterable:
            if await corofunc(item):
                yield item

alias(child, name)

Define an alias on our child.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required
name str

name of function tree

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/decorator.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def alias(child: CallableFunction, name: str) -> AsyncInnerFunction:
    """Define an alias on our child.

    Args:
        child (CallableFunction): child function to decorate
        name (str): name of function tree

    Returns:
        (AsyncInnerFunction): an awaitable function.
    """

    _child = to_async(child)

    # we use a dedicted function to 'duplicate' the child reference
    @node_metadata(name=name)
    async def _alias():
        return await _child()

    return _alias

always_failure(child)

Wrap child so the node always returns a falsy value.

If child returns a falsy result, that original result is preserved and returned. If child returns a truthy result, returns FAILURE instead. Exceptions are re-raised as ControlFlowException.

Note

To suppress exceptions as well, wrap child with ignore_exception first:

always_failure(child=ignore_exception(myfunction))

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to wrap.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the original falsy child result unchanged, or FAILURE if child is truthy.

Raises:

Type Description
ControlFlowException

if child raises an exception.
Source code in async_btree/decorator.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def always_failure(child: CallableFunction) -> AsyncInnerFunction:
    """Wrap child so the node always returns a falsy value.

    If child returns a falsy result, that original result is preserved and returned.
    If child returns a truthy result, returns `FAILURE` instead.
    Exceptions are re-raised as `ControlFlowException`.

    Note:
        To suppress exceptions as well, wrap child with `ignore_exception` first:

        `always_failure(child=ignore_exception(myfunction))`

    Args:
        child (CallableFunction): sync or async callable to wrap.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the original falsy child
            result unchanged, or `FAILURE` if child is truthy.

    Raises:
        ControlFlowException: if child raises an exception.
    """

    _child = to_async(child)

    @node_metadata()
    async def _always_failure():
        result: Any = FAILURE

        try:
            child_result = await _child()
            if not bool(child_result):
                result = child_result

        except Exception as e:
            raise ControlFlowException.instantiate(e) from e

        return result

    return _always_failure

always_success(child)

Wrap child so the node always returns a truthy value.

If child returns a truthy result, that original result is preserved and returned. If child returns a falsy result, returns SUCCESS instead. Exceptions are re-raised as ControlFlowException.

Note

To suppress exceptions as well, wrap child with ignore_exception first:

always_success(child=ignore_exception(myfunction))

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to wrap.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the original truthy child result unchanged, or SUCCESS if child is falsy.

Raises:

Type Description
ControlFlowException

if child raises an exception.
Source code in async_btree/decorator.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def always_success(child: CallableFunction) -> AsyncInnerFunction:
    """Wrap child so the node always returns a truthy value.

    If child returns a truthy result, that original result is preserved and returned.
    If child returns a falsy result, returns `SUCCESS` instead.
    Exceptions are re-raised as `ControlFlowException`.

    Note:
        To suppress exceptions as well, wrap child with `ignore_exception` first:

        `always_success(child=ignore_exception(myfunction))`

    Args:
        child (CallableFunction): sync or async callable to wrap.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the original truthy child
            result unchanged, or `SUCCESS` if child is falsy.

    Raises:
        ControlFlowException: if child raises an exception.
    """

    _child = to_async(child)

    @node_metadata()
    async def _always_success():
        result: Any = SUCCESS

        try:
            child_result = await _child()
            if bool(child_result):
                result = child_result

        except Exception as e:
            raise ControlFlowException.instantiate(e) from e

        return result

    return _always_success

amap(corofunc, iterable) async

Map an async function onto an iterable or an async iterable.

This simplify writing of mapping a function on something iterable between 'async for ...' and 'for...' .

Parameters:

Name Type Description Default
corofunc Callable[[Any], Awaitable[T]]

coroutine function

 required
iterable Union[AsyncIterable, Iterable]

iterable or async iterable collection which will be applied.

 required

Returns:

Type Description
AsyncGenerator[T, None]

AsyncGenerator[T]: an async iterator of corofunc(item)
Example

[i async for i in amap(inc, afilter(even, [0, 1, 2, 3, 4]))]

Source code in async_btree/utils.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
async def amap(corofunc: Callable[[Any], Awaitable[T]], iterable: AsyncIterable | Iterable) -> AsyncGenerator[T, None]:
    """Map an async function onto an iterable or an async iterable.

    This simplify writing of mapping a function on something iterable
    between 'async for ...' and 'for...' .

    Args:
        corofunc (Callable[[Any], Awaitable[T]]): coroutine function
        iterable (Union[AsyncIterable, Iterable]): iterable or async iterable collection
            which will be applied.

    Returns:
        AsyncGenerator[T]: an async iterator of corofunc(item)

    Example:
        ```[i async for i in amap(inc, afilter(even, [0, 1, 2, 3, 4]))]```

    """
    if isinstance(iterable, AsyncIterable):
        async for item in iterable:
            yield await corofunc(item)
    else:
        for item in iterable:
            yield await corofunc(item)

condition(target, **kwargs)

Declare a condition leaf node.

Delegates to is_success(action(target, **kwargs)) — returns SUCCESS if target is truthy, FAILURE otherwise. Exceptions from target propagate as ControlFlowException.

Parameters:

Name Type Description Default
target CallableFunction

sync or async callable evaluated as a boolean.

 required
kwargs

keyword arguments forwarded to target on each call.

 {}

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns SUCCESS or FAILURE.
Source code in async_btree/leaf.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def condition(target: CallableFunction, **kwargs) -> AsyncInnerFunction:
    """Declare a condition leaf node.

    Delegates to `is_success(action(target, **kwargs))` — returns `SUCCESS` if `target`
    is truthy, `FAILURE` otherwise. Exceptions from `target` propagate as
    `ControlFlowException`.

    Args:
        target (CallableFunction): sync or async callable evaluated as a boolean.
        kwargs: keyword arguments forwarded to `target` on each call.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns `SUCCESS` or `FAILURE`.
    """
    return alias_node_metadata(
        name="condition",
        target=is_success(action(target=target, **kwargs)),
        properties=["target"],
    )

condition_guard(condition, child)

Run child only if condition is truthy; return SUCCESS without running child if condition is falsy.

Semantic shorthand for decision(condition, success_tree=child) with no failure tree.

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable evaluated as a boolean.

 required
child CallableFunction

callable run only when condition is truthy.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns child result when condition is truthy, or SUCCESS when condition is falsy.
Source code in async_btree/control.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def condition_guard(condition: CallableFunction, child: CallableFunction) -> AsyncInnerFunction:
    """Run child only if condition is truthy; return SUCCESS without running child if condition is falsy.

    Semantic shorthand for `decision(condition, success_tree=child)` with no failure tree.

    Args:
        condition (CallableFunction): sync or async callable evaluated as a boolean.
        child (CallableFunction): callable run only when condition is truthy.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns child result when condition
            is truthy, or `SUCCESS` when condition is falsy.
    """
    return alias_node_metadata(name="condition_guard", target=decision(condition=condition, success_tree=child))

cooldown(child, delay, throttled_value=SUCCESS)

Skip child if called again before delay seconds have elapsed since the last run.

Last-run time is stored in the closure — it persists across BTreeRunner.run() ticks for the lifetime of this node instance.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to throttle.

 required
delay float

minimum seconds between successive executions of child.

 required
throttled_value Any

value returned when child is skipped. Defaults to SUCCESS.

 SUCCESS

Returns:

Type Description
AsyncInnerFunction

an awaitable function that runs child and returns its result when the cooldown has elapsed, or throttled_value when the call is throttled.
Source code in async_btree/decorator.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def cooldown(child: CallableFunction, delay: float, throttled_value: Any = SUCCESS) -> AsyncInnerFunction:
    """Skip child if called again before `delay` seconds have elapsed since the last run.

    Last-run time is stored in the closure — it persists across `BTreeRunner.run()` ticks
    for the lifetime of this node instance.

    Args:
        child (CallableFunction): sync or async callable to throttle.
        delay (float): minimum seconds between successive executions of child.
        throttled_value (Any): value returned when child is skipped. Defaults to `SUCCESS`.

    Returns:
        (AsyncInnerFunction): an awaitable function that runs child and returns its result
            when the cooldown has elapsed, or `throttled_value` when the call is throttled.
    """
    _child = to_async(child)
    _last_run: list[float] = [0.0]  # list to allow mutation from inner scope

    @node_metadata(properties=["delay"])
    async def _cooldown() -> Any:
        now = time.monotonic()
        if now - _last_run[0] < delay:
            return throttled_value
        _last_run[0] = now
        return await _child()

    return _cooldown

decision(condition, success_tree, failure_tree=None)

Create a decision node.

If condition is truthy, return evaluation of success_tree. Otherwise, return evaluation of failure_tree if set, or SUCCESS.

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable evaluated as a boolean.

 required
success_tree CallableFunction

callable evaluated when condition is truthy.

 required
failure_tree CallableFunction | None

callable evaluated when condition is falsy. Defaults to None; when absent, returns SUCCESS.

 None

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/control.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def decision(
    condition: CallableFunction,
    success_tree: CallableFunction,
    failure_tree: CallableFunction | None = None,
) -> AsyncInnerFunction:
    """Create a decision node.

    If condition is truthy, return evaluation of `success_tree`.
    Otherwise, return evaluation of `failure_tree` if set, or `SUCCESS`.

    Args:
        condition (CallableFunction): sync or async callable evaluated as a boolean.
        success_tree (CallableFunction): callable evaluated when condition is truthy.
        failure_tree (CallableFunction | None): callable evaluated when condition is falsy.
            Defaults to `None`; when absent, returns `SUCCESS`.

    Returns:
        (AsyncInnerFunction): an awaitable function.
    """

    _condition = to_async(condition)
    _success_tree = to_async(success_tree)
    _failure_tree = to_async(failure_tree) if failure_tree else None

    @node_metadata(edges=["_condition", "_success_tree", "_failure_tree"])
    async def _decision():
        if bool(await _condition()):
            return await _success_tree()
        if _failure_tree:
            return await _failure_tree()
        return SUCCESS

    return _decision

decorate(child, decorator, **kwargs)

Post-process child result with a decorator function.

Runs child eagerly, then passes the result as the first argument to decorator.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to run first.

 required
decorator CallableFunction

sync or async callable with signature decorator(child_result, **kwargs).

 required
kwargs

additional keyword arguments forwarded to decorator.

 {}

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns decorator(child_result, **kwargs).
Source code in async_btree/decorator.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def decorate(child: CallableFunction, decorator: CallableFunction, **kwargs) -> AsyncInnerFunction:
    """Post-process child result with a decorator function.

    Runs child eagerly, then passes the result as the first argument to `decorator`.

    Args:
        child (CallableFunction): sync or async callable to run first.
        decorator (CallableFunction): sync or async callable with signature
            `decorator(child_result, **kwargs)`.
        kwargs: additional keyword arguments forwarded to `decorator`.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns `decorator(child_result, **kwargs)`.
    """

    _child = to_async(child)
    _decorator = to_async(decorator)

    @node_metadata(properties=["_decorator"])
    async def _decorate():
        return await _decorator(await _child(), **kwargs)

    return _decorate

delay(child, seconds)

Wait seconds before running child.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to run after the delay.

 required
seconds float

number of seconds to wait before executing child.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that sleeps then returns child result.
Source code in async_btree/decorator.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
def delay(child: CallableFunction, seconds: float) -> AsyncInnerFunction:
    """Wait `seconds` before running child.

    Args:
        child (CallableFunction): sync or async callable to run after the delay.
        seconds (float): number of seconds to wait before executing child.

    Returns:
        (AsyncInnerFunction): an awaitable function that sleeps then returns child result.
    """
    _child = to_async(child)

    @node_metadata(properties=["seconds"])
    async def _delay():
        await anyio.sleep(seconds)
        return await _child()

    return _delay

do_while(child, condition)

Run child at least once, then repeat while condition is truthy.

Unlike repeat_while, the child always executes at least once before the condition is checked.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable executed each iteration.

 required
condition CallableFunction

sync or async callable checked after each iteration.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the last child result.
Source code in async_btree/control.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def do_while(child: CallableFunction, condition: CallableFunction) -> AsyncInnerFunction:
    """Run child at least once, then repeat while condition is truthy.

    Unlike `repeat_while`, the child always executes at least once before the condition
    is checked.

    Args:
        child (CallableFunction): sync or async callable executed each iteration.
        condition (CallableFunction): sync or async callable checked after each iteration.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the last child result.
    """
    _child = to_async(child)
    _condition = to_async(condition)

    @node_metadata(edges=["_child", "_condition"])
    async def _do_while():
        result: Any = FAILURE
        while True:
            result = await _child()
            if not bool(await _condition()):
                break
        return result

    return _do_while

fallback(children)

Execute children in sequence and succeed as soon as one succeeds; fail if all fail.

Children are evaluated in order from highest to lowest priority. Evaluation stops at the first truthy result. selector is an alias for this function.

Parameters:

Name Type Description Default
children list[CallableFunction]

list of sync or async callables.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the first truthy result, or the last falsy result if all children fail.
Source code in async_btree/control.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def fallback(children: list[CallableFunction]) -> AsyncInnerFunction:
    """Execute children in sequence and succeed as soon as one succeeds; fail if all fail.

    Children are evaluated in order from highest to lowest priority. Evaluation stops
    at the first truthy result. `selector` is an alias for this function.

    Args:
        children (list[CallableFunction]): list of sync or async callables.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the first truthy result,
            or the last falsy result if all children fail.
    """
    return alias_node_metadata(
        name="fallback",
        target=sequence(children, success_threshold=min(1, len(children))),
    )

ignore_exception(child)

Wrap child so exceptions are caught and returned as a falsy value instead of propagating.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to wrap.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns child result unchanged on success, or a ControlFlowException wrapping the exception on failure. The returned exception is falsy.
Source code in async_btree/decorator.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def ignore_exception(child: CallableFunction) -> AsyncInnerFunction:
    """Wrap child so exceptions are caught and returned as a falsy value instead of propagating.

    Args:
        child (CallableFunction): sync or async callable to wrap.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns child result unchanged on
            success, or a `ControlFlowException` wrapping the exception on failure.
            The returned exception is falsy.
    """

    _child = to_async(child)

    @node_metadata()
    async def _ignore_exception():
        try:
            return await _child()

        except Exception as e:
            return ControlFlowException.instantiate(e)

    return _ignore_exception

inverter(child)

Invert node status.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to invert.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns SUCCESS if child is falsy, else FAILURE.
Source code in async_btree/decorator.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def inverter(child: CallableFunction) -> AsyncInnerFunction:
    """Invert node status.

    Args:
        child (CallableFunction): sync or async callable to invert.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns `SUCCESS` if child
            is falsy, else `FAILURE`.
    """

    _child = to_async(child)

    @node_metadata()
    async def _inverter():
        return not bool(await _child())

    return _inverter

is_failure(child)

Return SUCCESS if child is falsy, FAILURE otherwise.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to test.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns SUCCESS if child result is falsy, FAILURE if child result is truthy.
Source code in async_btree/decorator.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def is_failure(child: CallableFunction) -> AsyncInnerFunction:
    """Return `SUCCESS` if child is falsy, `FAILURE` otherwise.

    Args:
        child (CallableFunction): sync or async callable to test.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns `SUCCESS` if child
            result is falsy, `FAILURE` if child result is truthy.
    """

    _child = to_async(child)

    @node_metadata()
    async def _is_failure():
        return SUCCESS if not bool(await _child()) else FAILURE

    return _is_failure

is_success(child)

Create a conditional node which test if child success.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which return SUCCESS if child return SUCCESS else FAILURE.
Source code in async_btree/decorator.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def is_success(child: CallableFunction) -> AsyncInnerFunction:
    """Create a conditional node which test if child success.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which return SUCCESS if child
            return SUCCESS else FAILURE.
    """

    _child = to_async(child)

    @node_metadata()
    async def _is_success():
        return SUCCESS if bool(await _child()) else FAILURE

    return _is_success

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 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
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

parallel_race(children)

Run children concurrently; return the result of the first child to finish.

As soon as one child completes, all remaining children are cancelled. Returns FAILURE if children is empty.

Parameters:

Name Type Description Default
children list[CallableFunction]

list of sync or async callables.

 required

Returns:

Name Type Description
AsyncInnerFunction AsyncInnerFunction

an awaitable function that returns the winner's result.

Raises:

Type Description
ControlFlowException

wrapping any exception raised by a child.
Source code in async_btree/parallele.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def parallel_race(children: list[CallableFunction]) -> AsyncInnerFunction:
    """Run children concurrently; return the result of the first child to finish.

    As soon as one child completes, all remaining children are cancelled.
    Returns `FAILURE` if `children` is empty.

    Args:
        children: list of sync or async callables.

    Returns:
        AsyncInnerFunction: an awaitable function that returns the winner's result.

    Raises:
        ControlFlowException: wrapping any exception raised by a child.
    """
    _children = [to_async(child) for child in children]

    @node_metadata()
    async def _parallel_race() -> Any:
        winner: list[Any] = []

        async def _run(child: AsyncInnerFunction) -> None:
            result = await child()
            if not winner:
                winner.append(result)
                tg.cancel_scope.cancel()

        try:
            async with anyio.create_task_group() as tg:
                for child in _children:
                    tg.start_soon(_run, child)
        except Exception as e:
            raise ControlFlowException.instantiate(e) from e

        return winner[0] if winner else FAILURE

    return _parallel_race

random_selector(children)

Execute children in a random order, succeeding as soon as one succeeds.

Children are reshuffled on every call, so the evaluation order differs each tick.

Parameters:

Name Type Description Default
children list[CallableFunction]

list of sync or async callables.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns [result] for the first truthy child, or FAILURE if all children fail.
Source code in async_btree/control.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def random_selector(children: list[CallableFunction]) -> AsyncInnerFunction:
    """Execute children in a random order, succeeding as soon as one succeeds.

    Children are reshuffled on every call, so the evaluation order differs each tick.

    Args:
        children (list[CallableFunction]): list of sync or async callables.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns `[result]` for the first
            truthy child, or `FAILURE` if all children fail.
    """
    _children = [to_async(child) for child in children]

    @node_metadata()
    async def _random_selector():
        for child in random.sample(_children, len(_children)):
            result = await child()
            if bool(result):
                return [result]
        return FAILURE

    return _random_selector

repeat_n(child, n)

Run child exactly n times regardless of its result.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to run.

 required
n int

number of times to execute child.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the last child result, or FAILURE if n is 0.
Source code in async_btree/control.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def repeat_n(child: CallableFunction, n: int) -> AsyncInnerFunction:
    """Run child exactly `n` times regardless of its result.

    Args:
        child (CallableFunction): sync or async callable to run.
        n (int): number of times to execute child.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the last child result,
            or `FAILURE` if `n` is 0.
    """
    _child = to_async(child)

    @node_metadata(properties=["n"])
    async def _repeat_n():
        result: Any = FAILURE
        for _ in range(n):
            result = await _child()
        return result

    return _repeat_n

repeat_until(condition, child)

Repeat child until condition becomes truthy, stopping when condition is met.

Returns the last child evaluation, or FAILURE if condition is truthy on the first check (no iteration occurs).

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable evaluated before each iteration.

 required
child CallableFunction

sync or async callable executed each iteration.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/control.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def repeat_until(condition: CallableFunction, child: CallableFunction) -> AsyncInnerFunction:
    """Repeat child until condition becomes truthy, stopping when condition is met.

    Returns the last child evaluation, or `FAILURE` if condition is truthy on the first check
    (no iteration occurs).

    Args:
        condition (CallableFunction): sync or async callable evaluated before each iteration.
        child (CallableFunction): sync or async callable executed each iteration.

    Returns:
        (AsyncInnerFunction): an awaitable function.
    """

    _child = to_async(child)
    _condition = to_async(condition)

    @node_metadata(edges=["_condition", "_child"])
    async def _repeat_until():
        result: Any = FAILURE
        while not bool(await _condition()):
            result = await _child()

        return result

    return _repeat_until

repeat_while(condition, child)

Repeat child while condition is truthy, stopping when condition becomes falsy.

Returns the last child evaluation, or FAILURE if condition is falsy on the first check.

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable evaluated before each iteration.

 required
child CallableFunction

sync or async callable executed each iteration.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function.
Source code in async_btree/control.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def repeat_while(condition: CallableFunction, child: CallableFunction) -> AsyncInnerFunction:
    """Repeat child while condition is truthy, stopping when condition becomes falsy.

    Returns the last child evaluation, or `FAILURE` if condition is falsy on the first check.

    Args:
        condition (CallableFunction): sync or async callable evaluated before each iteration.
        child (CallableFunction): sync or async callable executed each iteration.

    Returns:
        (AsyncInnerFunction): an awaitable function.
    """

    _child = to_async(child)
    _condition = to_async(condition)

    @node_metadata(edges=["_condition", "_child"])
    async def _repeat_while():
        result: Any = FAILURE
        while bool(await _condition()):
            result = await _child()

        return result

    return _repeat_while

retry(child, max_retry=3)

Retry child evaluation on failure until it succeeds or max_retry attempts are exhausted.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to retry.

 required
max_retry int

maximum number of attempts (default 3). Use -1 for infinite retries.

 3

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the first truthy result, or the last falsy result (which may be a ControlFlowException) if all attempts fail.

Raises:

Type Description
AssertionError

if max_retry is 0 or negative (other than -1).
Source code in async_btree/decorator.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def retry(child: CallableFunction, max_retry: int = 3) -> AsyncInnerFunction:
    """Retry child evaluation on failure until it succeeds or `max_retry` attempts are exhausted.

    Args:
        child (CallableFunction): sync or async callable to retry.
        max_retry (int): maximum number of attempts (default 3). Use `-1` for infinite retries.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the first truthy result, or
            the last falsy result (which may be a `ControlFlowException`) if all attempts fail.

    Raises:
        AssertionError: if `max_retry` is 0 or negative (other than -1).
    """
    if not (max_retry > 0 or max_retry == -1):
        raise AssertionError("max_retry")

    _child = to_async(child)

    @node_metadata(properties=["max_retry"])
    async def _retry():
        retry_count = max_retry
        result: Any = FAILURE

        while not bool(result) and retry_count != 0:
            result = await _child()
            retry_count -= 1

        return result

    return _retry

retry_until_failed(child)

Retry child until failed.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which try to evaluate child until it failed.
Source code in async_btree/decorator.py 
298
299
300
301
302
303
304
305
306
307
308
309
def retry_until_failed(child: CallableFunction) -> AsyncInnerFunction:
    """Retry child until failed.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which try to evaluate child
            until it failed.
    """

    return alias_node_metadata(name="retry_until_failed", target=retry(child=inverter(child), max_retry=-1))

retry_until_success(child)

Retry child until success.

Parameters:

Name Type Description Default
child CallableFunction

child function to decorate

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function which try to evaluate child until it succeed.
Source code in async_btree/decorator.py 
285
286
287
288
289
290
291
292
293
294
295
def retry_until_success(child: CallableFunction) -> AsyncInnerFunction:
    """Retry child until success.

    Args:
        child (CallableFunction): child function to decorate

    Returns:
        (AsyncInnerFunction): an awaitable function which try to evaluate child
            until it succeed.
    """
    return alias_node_metadata(name="retry_until_success", target=retry(child=child, max_retry=-1))

run(target, *args, backend='asyncio', **kwargs)

Run a behavior tree callable to completion.

Convenience wrapper around BTreeRunner for one-shot execution.

Parameters:

Name Type Description Default
target Callable[..., Awaitable[Any]]

async callable (coroutine function)

 required
*args Any

positional arguments passed to target

 ()
backend Backend

async runtime — "asyncio" (default), "trio", or "asyncio+uvloop"

 'asyncio'
**kwargs Any

keyword arguments passed to target

 {}

Returns:

Type Description
Any

whatever target returns
Source code in async_btree/utils.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def run(
    target: Callable[..., Awaitable[Any]],
    *args: Any,
    backend: Backend = "asyncio",
    **kwargs: Any,
) -> Any:
    """Run a behavior tree callable to completion.

    Convenience wrapper around BTreeRunner for one-shot execution.

    Args:
        target: async callable (coroutine function)
        *args: positional arguments passed to target
        backend: async runtime — "asyncio" (default), "trio", or "asyncio+uvloop"
        **kwargs: keyword arguments passed to target

    Returns:
        whatever target returns
    """
    with BTreeRunner(backend=backend) as runner:
        return runner.run(target, *args, **kwargs)

selector(children)

Alias of fallback. Execute children in sequence and succeed as soon as one succeeds.

Children are evaluated in order from highest to lowest priority. Evaluation stops at the first truthy result.

Parameters:

Name Type Description Default
children list[CallableFunction]

list of sync or async callables.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the first truthy result, or the last falsy result if all children fail.
Source code in async_btree/control.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def selector(children: list[CallableFunction]) -> AsyncInnerFunction:
    """Alias of `fallback`. Execute children in sequence and succeed as soon as one succeeds.

    Children are evaluated in order from highest to lowest priority. Evaluation stops
    at the first truthy result.

    Args:
        children (list[CallableFunction]): list of sync or async callables.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the first truthy result,
            or the last falsy result if all children fail.
    """
    return alias_node_metadata(
        name="selector",
        target=sequence(children, success_threshold=min(1, len(children))),
    )

sequence(children, success_threshold=None)

Return a function which executes children in sequence.

success_threshold generalizes traditional sequence/fallback behaviour and must be in [0, len(children)]. Defaults to None, which means len(children) (all children must succeed).

  • If #success reaches success_threshold, returns the list of results collected so far.
  • If #failure reaches len(children) - success_threshold + 1, returns the last falsy result.

Parameters:

Name Type Description Default
children list[CallableFunction]

list of sync or async callables.

 required
success_threshold int | None

minimum number of children that must succeed. Defaults to None (equivalent to len(children)).

 None

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns a list of results on success, or the last falsy result on failure.

Raises:

Type Description
AssertionError

if success_threshold is outside [0, len(children)].
Source code in async_btree/control.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def sequence(children: list[CallableFunction], success_threshold: int | None = None) -> AsyncInnerFunction:
    """Return a function which executes children in sequence.

    `success_threshold` generalizes traditional sequence/fallback behaviour and
    must be in `[0, len(children)]`. Defaults to `None`, which means `len(children)`
    (all children must succeed).

    - If #success reaches `success_threshold`, returns the list of results collected so far.
    - If #failure reaches `len(children) - success_threshold + 1`, returns the last falsy result.

    Args:
        children (list[CallableFunction]): list of sync or async callables.
        success_threshold (int | None): minimum number of children that must succeed.
            Defaults to `None` (equivalent to `len(children)`).

    Returns:
        (AsyncInnerFunction): an awaitable function that returns a list of results on
            success, or the last falsy result on failure.

    Raises:
        AssertionError: if `success_threshold` is outside `[0, len(children)]`.
    """
    _success_threshold = success_threshold if success_threshold is not None else len(children)
    if not (0 <= _success_threshold <= len(children)):
        raise AssertionError("success_threshold")

    failure_threshold = len(children) - _success_threshold + 1

    _children = [to_async(child) for child in children]

    @node_metadata(properties=["_success_threshold"])
    async def _sequence():
        success = 0
        failure = 0
        results = []

        for child in _children:
            last_result = await child()
            results.append(last_result)

            if bool(last_result):
                success += 1
                if success == _success_threshold:
                    return results
            else:
                failure += 1
                if failure == failure_threshold:
                    return last_result
        return FAILURE

    return _sequence

stringify_analyze(target, indent=0, label=None)

Stringify a Node tree into a human-readable indented representation.

Parameters:

Name Type Description Default
target Node

node to stringify.

 required
indent int

current indentation level (default 0).

 0
label Optional[str]

edge label to prefix the node with (default None).

 None

Returns:

Type Description
str

indented string representation of the node tree.
Source code in async_btree/analyze.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def stringify_analyze(target: Node, indent: int = 0, label: str | None = None) -> str:
    """Stringify a `Node` tree into a human-readable indented representation.

    Args:
        target (Node): node to stringify.
        indent (int): current indentation level (default 0).
        label (Optional[str]): edge label to prefix the node with (default `None`).

    Returns:
        (str): indented string representation of the node tree.
    """
    _ident = "    "
    _space = f"{_ident * indent} "
    result: str = ""
    if label:
        result += f"{_space}--({label})--> {target.name}:\n"
        _space += f"{_ident}{' ' * len(label)}"
    else:
        result += f"{_space}--> {target.name}:\n"

    for k, v in target.properties:
        result += f"{_space}    {k}: {v}\n"

    for _label, children in target.edges:
        if children:
            for child in children:
                result += stringify_analyze(target=child, indent=indent + 1, label=_label)
    return result

switch(condition, cases, default=None)

Route to a child based on the return value of condition.

Evaluates condition to get a key, looks it up in cases, and runs the matching child. Falls back to default if no case matches. Returns FAILURE if no case matches and no default is provided.

The case keys are shown in stringify_analyze output. The children themselves are not traversed by analyze().

Parameters:

Name Type Description Default
condition CallableFunction

sync or async callable that returns a hashable key.

 required
cases dict[Any, CallableFunction]

mapping of key → child callable.

 required
default CallableFunction | None

callable run when no case matches. Defaults to None.

 None

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns the matched child's result, or FAILURE if no match and no default.
Source code in async_btree/control.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
def switch(
    condition: CallableFunction,
    cases: dict[Any, CallableFunction],
    default: CallableFunction | None = None,
) -> AsyncInnerFunction:
    """Route to a child based on the return value of `condition`.

    Evaluates `condition` to get a key, looks it up in `cases`, and runs the
    matching child. Falls back to `default` if no case matches. Returns `FAILURE`
    if no case matches and no default is provided.

    The case keys are shown in `stringify_analyze` output. The children themselves
    are not traversed by `analyze()`.

    Args:
        condition (CallableFunction): sync or async callable that returns a hashable key.
        cases (dict[Any, CallableFunction]): mapping of key → child callable.
        default (CallableFunction | None): callable run when no case matches. Defaults to `None`.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns the matched child's result,
            or `FAILURE` if no match and no default.
    """
    _condition = to_async(condition)
    _cases = {k: to_async(v) for k, v in cases.items()}
    _default = to_async(default) if default else None
    _case_keys = list(cases.keys())

    @node_metadata(properties=["_case_keys"], edges=["_default"])
    async def _switch():
        _ = _case_keys  # ensure captured in closure for analyze()
        key = await _condition()
        child = _cases.get(key, _default)
        if child is None:
            return FAILURE
        return await child()

    return _switch

timeout_after(child, delay)

Run child with a time limit; return FAILURE if the deadline is exceeded.

Parameters:

Name Type Description Default
child CallableFunction

sync or async callable to run.

 required
delay float

maximum seconds to wait before returning FAILURE.

 required

Returns:

Type Description
AsyncInnerFunction

an awaitable function that returns child result on success, or FAILURE if the deadline is exceeded.
Source code in async_btree/decorator.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
def timeout_after(child: CallableFunction, delay: float) -> AsyncInnerFunction:
    """Run child with a time limit; return `FAILURE` if the deadline is exceeded.

    Args:
        child (CallableFunction): sync or async callable to run.
        delay (float): maximum seconds to wait before returning `FAILURE`.

    Returns:
        (AsyncInnerFunction): an awaitable function that returns child result on
            success, or `FAILURE` if the deadline is exceeded.
    """
    _child = to_async(child)

    @node_metadata(properties=["delay"])
    async def _timeout_after():
        result: Any = FAILURE
        with anyio.move_on_after(delay) as cancel_scope:
            result = await _child()
        if cancel_scope.cancelled_caught:
            return FAILURE
        return result

    return _timeout_after