Coverage for promplate/chain/utils.py: 84%
29 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-08-08 04:34 +0800
« prev ^ index » next coverage.py v7.6.1, created at 2024-08-08 04:34 +0800
1from inspect import Parameter, isawaitable, signature
2from itertools import accumulate
3from typing import AsyncIterable, Awaitable, Callable, Iterable, TypeVar, cast
5T = TypeVar("T")
8def appender(to_append: list[T]) -> Callable[[T], T]:
9 def append_processer(func: T) -> T:
10 to_append.append(func)
12 return func
14 return append_processer
17def is_positional_parameter(p: Parameter):
18 return p.kind is Parameter.POSITIONAL_OR_KEYWORD or p.kind is Parameter.KEYWORD_ONLY
21def count_position_parameters(func):
22 return sum(map(is_positional_parameter, signature(func).parameters.values()))
25async def resolve(maybe_awaitable: T | Awaitable[T], /) -> T:
26 while isawaitable(maybe_awaitable):
27 maybe_awaitable = await maybe_awaitable
29 return maybe_awaitable # type: ignore
32async def async_accumulate(async_iterable: AsyncIterable[str]):
33 result = ""
34 async for delta in async_iterable:
35 result += delta
36 yield result
39def accumulate_any(any_iterable: Iterable[str] | AsyncIterable[str]):
40 if "__aiter__" in dir(any_iterable):
41 return async_accumulate(any_iterable) # type: ignore
43 async def _():
44 for i in accumulate(cast(Iterable[str], any_iterable)):
45 yield i
47 return _()