Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/ijson/adapters.py: 24%

1from ijson import compat 

2 

3try: 

4 from builtins import aiter, anext 

5except ImportError: # Py<3.10 

6 _MISSING = object() 

7 

8 def aiter(obj): 

9 return obj.__aiter__() 

10 

11 async def anext(ait, default=_MISSING): 

12 try: 

13 return await ait.__anext__() 

14 except StopAsyncIteration: 

15 if default is _MISSING: 

16 raise 

17 return default 

18 

19 

20def _to_bytes(chunk, warned: bool): 

21 if isinstance(chunk, bytes): 

22 return chunk, warned 

23 if isinstance(chunk, str): 

24 if not warned: 

25 compat._warn_and_return(None) 

26 warned = True 

27 return chunk.encode("utf-8"), warned 

28 raise TypeError("from_iter expects an iterable of bytes or str") 

29 

30 

31class IterReader: 

32 """File-like object backed by a byte iterator.""" 

33 

34 def __init__(self, byte_iter): 

35 self._iter = byte_iter 

36 self._warned = False 

37 

38 def read(self, n: int) -> bytes: 

39 if n == 0: 

40 return b"" 

41 chunk, self._warned = _to_bytes(next(self._iter, b""), self._warned) 

42 return chunk 

43 

44 

45class AiterReader: 

46 """Async file-like object backed by an async byte iterator.""" 

47 

48 def __init__(self, byte_aiter): 

49 self._aiter = byte_aiter 

50 self._warned = False 

51 

52 async def read(self, n: int) -> bytes: 

53 if n == 0: 

54 return b"" 

55 chunk, self._warned = _to_bytes(await anext(self._aiter, b""), self._warned) 

56 return chunk 

57 

58 

59def from_iter(byte_iter): 

60 """Convert a byte iterable (sync or async) to a file-like object.""" 

61 if hasattr(byte_iter, "__aiter__"): 

62 return AiterReader(aiter(byte_iter)) 

63 return IterReader(iter(byte_iter))