Coverage for promplate/chain/utils.py: 84%

1from inspect import Parameter, isawaitable, signature 

2from itertools import accumulate 

3from typing import AsyncIterable, Awaitable, Callable, Iterable, TypeVar, cast 

4 

5T = TypeVar("T") 

6 

7 

8def appender(to_append: list[T]) -> Callable[[T], T]: 

9 def append_processer(func: T) -> T: 

10 to_append.append(func) 

11 

12 return func 

13 

14 return append_processer 

15 

16 

17def is_positional_parameter(p: Parameter): 

18 return p.kind is Parameter.POSITIONAL_OR_KEYWORD or p.kind is Parameter.KEYWORD_ONLY 

19 

20 

21def count_position_parameters(func): 

22 return sum(map(is_positional_parameter, signature(func).parameters.values())) 

23 

24 

25async def resolve(maybe_awaitable: T | Awaitable[T], /) -> T: 

26 while isawaitable(maybe_awaitable): 

27 maybe_awaitable = await maybe_awaitable 

28 

29 return maybe_awaitable # type: ignore 

30 

31 

32async def async_accumulate(async_iterable: AsyncIterable[str]): 

33 result = "" 

34 async for delta in async_iterable: 

35 result += delta 

36 yield result 

37 

38 

39def accumulate_any(any_iterable: Iterable[str] | AsyncIterable[str]): 

40 if "__aiter__" in dir(any_iterable): 

41 return async_accumulate(any_iterable) # type: ignore 

42 

43 async def _(): 

44 for i in accumulate(cast(Iterable[str], any_iterable)): 

45 yield i 

46 

47 return _()