Skip to content

utils

async_btree.utils

Utility function.

Attributes

Classes

Functions

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

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)

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)

run_once(target)

Implement 'run once' function.

The target function is called exactly once. Any further call will return the first result. This decorator works on async and sync functions.

Parameters:

Name Type Description Default
target CallableFunction

target function

 required

Returns:

Name Type Description
CallableFunction CallableFunction

decorated run once function.
Source code in async_btree/utils.py 
 96
 97
 98
 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
def run_once(target: CallableFunction) -> CallableFunction:
    """Implement 'run once' function.

    The target function is called exactly once. Any further call will return the first result.
    This decorator works on async and sync functions.

    Args:
        target (CallableFunction): target function

    Returns:
        CallableFunction: decorated run once function.
    """
    _result = None
    _has_run = False

    if not iscoroutinefunction(target):

        @wraps(target)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            nonlocal _result, _has_run
            if not _has_run:
                _has_run = True
                _result = target(*args, **kwargs)
            return _result

        return sync_wrapper

    async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
        nonlocal _result, _has_run
        if not _has_run:
            _has_run = True
            _result = await target(*args, **kwargs)  # type: ignore[misc]
        return _result

    return async_wrapper

to_async(target)

Return target unchanged if already async, otherwise wrap it in an async function.

The returned wrapper carries __node_metadata with the original function's name, so sync functions participate in tree introspection via analyze().

Parameters:

Name Type Description Default
target CallableFunction

sync or async callable.

 required

Returns:

Type Description
Callable[..., Awaitable[Any]]

an async version of target.
Source code in async_btree/utils.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def to_async(target: CallableFunction) -> Callable[..., Awaitable[Any]]:
    """Return `target` unchanged if already async, otherwise wrap it in an async function.

    The returned wrapper carries `__node_metadata` with the original function's name,
    so sync functions participate in tree introspection via `analyze()`.

    Args:
        target (CallableFunction): sync or async callable.

    Returns:
        (Callable[..., Awaitable[Any]]): an async version of `target`.
    """
    if iscoroutinefunction(target):
        return target

    @node_metadata(name=target.__name__.lstrip("_") if hasattr(target, "__name__") else "anonymous")
    async def _to_async(*args: Any, **kwargs: Any) -> Any:
        return target(*args, **kwargs)

    return _to_async