python/jc
1
0
Fork 0
mirror of https://github.com/kellyjonbrazil/jc.git synced 2025-07-13 01:20:24 +02:00
Code Issues Releases Activity

type annotate decorator function

Browse Source
This commit is contained in:
Kelly Brazil
2022-10-19 12:02:12 -07:00
parent af7c4ce8ec
commit 1559c4751e
1 changed files with 6 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
jc/streaming.py
View File

@ -1,10 +1,13 @@
"""jc - JSON Convert streaming utils""" """jc - JSON Convert streaming utils"""
from functools import wraps from functools import wraps
from typing import Dict, Tuple, Union, Iterable from typing import Dict, Tuple, Union, Iterable, Callable, TypeVar, cast, Any
from .jc_types import JSONDictType, MetadataType from .jc_types import JSONDictType, MetadataType
F = TypeVar('F', bound=Callable[..., Any])
def streaming_input_type_check(data: Iterable[Union[str, bytes]]) -> None: def streaming_input_type_check(data: Iterable[Union[str, bytes]]) -> None:
""" """
Ensure input data is an iterable, but not a string or bytes. Raises Ensure input data is an iterable, but not a string or bytes. Raises
@ -42,7 +45,7 @@ def stream_error(e: BaseException, line: str) -> Dict[str, MetadataType]:
} }
def add_jc_meta(func): def add_jc_meta(func: F) -> F:
""" """
Decorator for streaming parsers to add stream_success and stream_error Decorator for streaming parsers to add stream_success and stream_error
objects. This simplifies the yield lines in the streaming parsers. objects. This simplifies the yield lines in the streaming parsers.
@ -97,7 +100,7 @@ def add_jc_meta(func):
line = value[1] line = value[1]
yield stream_error(exception_obj, line) yield stream_error(exception_obj, line)
return wrapper return cast(F, wrapper)
def raise_or_yield( def raise_or_yield(