From dfa7a71f5348f3061c9286c0ad90b226169eb8c5 Mon Sep 17 00:00:00 2001 From: Kelly Brazil Date: Tue, 18 Oct 2022 11:01:59 -0700 Subject: [PATCH] move custom types to jc_types --- jc/cli.py | 12 ++++------ jc/jc_types.py | 55 +++++++++++++++++++++++++++++++++++++++++++++ jc/lib.py | 28 ++--------------------- jc/parsers/foo.py | 9 ++++---- jc/parsers/foo_s.py | 5 +++-- jc/streaming.py | 11 ++++----- jc/utils.py | 18 +++------------ 7 files changed, 78 insertions(+), 60 deletions(-) create mode 100644 jc/jc_types.py diff --git a/jc/cli.py b/jc/cli.py index e6eefa3b..d8cd6bdb 100644 --- a/jc/cli.py +++ b/jc/cli.py @@ -10,13 +10,13 @@ import textwrap import signal import shlex import subprocess -from typing import List, Dict, Union, Optional, TextIO +from typing import List, Union, Optional, TextIO from types import ModuleType from .lib import ( __version__, parser_info, all_parser_info, parsers, _get_parser, _parser_is_streaming, - parser_mod_list, standard_parser_mod_list, plugin_parser_mod_list, streaming_parser_mod_list, - JSONDictType, ParserInfoType + parser_mod_list, standard_parser_mod_list, plugin_parser_mod_list, streaming_parser_mod_list ) +from .jc_types import JSONDictType, AboutJCType, MetadataType, CustomColorType from . import utils from .cli_data import ( long_options_map, new_pygments_colors, old_pygments_colors, helptext_preamble_string, @@ -26,9 +26,6 @@ from .shell_completions import bash_completion, zsh_completion from . import tracebackplus from .exceptions import LibraryNotInstalled, ParseError -MetadataType = Dict[str, Optional[Union[str, int, float, List[str], datetime]]] -AboutJCType = Dict[str, Union[str, int, List[ParserInfoType]]] - # make pygments import optional PYGMENTS_INSTALLED: bool = False try: @@ -39,9 +36,8 @@ try: from pygments.lexers.data import JsonLexer, YamlLexer from pygments.formatters import Terminal256Formatter PYGMENTS_INSTALLED = True - CustomColorType = Dict[Union[Name.Tag, Number, String, Keyword], str] except Exception: - CustomColorType = Dict # type: ignore + pass JC_CLEAN_EXIT: int = 0 JC_ERROR_EXIT: int = 100 diff --git a/jc/jc_types.py b/jc/jc_types.py new file mode 100644 index 00000000..0597165f --- /dev/null +++ b/jc/jc_types.py @@ -0,0 +1,55 @@ +"""jc - JSON Convert lib module""" + +import sys +from datetime import datetime +from typing import Dict, List, Iterator, Optional, Union + +JSONDictType = Dict[str, Union[str, int, float, bool, List, Dict, None]] +MetadataType = Dict[str, Optional[Union[str, int, float, List[str], datetime]]] +StreamingOutputType = Iterator[Union[JSONDictType, tuple[BaseException, str]]] + +if sys.version_info >= (3, 8): + from typing import TypedDict + + ParserInfoType = TypedDict( + 'ParserInfoType', + { + "name": str, + "argument": str, + "version": str, + "description": str, + "author": str, + "author_email": str, + "compatible": List[str], + "magic_commands": List[str], + "documentation": str, + "streaming": bool, + "plugin": bool, + "hidden": bool, + "deprecated": bool + }, + total=False + ) + + TimeStampFormatType = TypedDict( + 'TimeStampFormatType', + { + 'id': int, + 'format': str, + 'locale': Optional[str] + } + ) + +else: + ParserInfoType = Dict + TimeStampFormatType = Dict + +AboutJCType = Dict[str, Union[str, int, List[ParserInfoType]]] + + +try: + from pygments.token import (Name, Number, String, Keyword) + CustomColorType = Dict[Union[Name.Tag, Number, String, Keyword], str] + +except Exception: + CustomColorType = Dict # type: ignore diff --git a/jc/lib.py b/jc/lib.py index 35ad6a76..ebcff0a0 100644 --- a/jc/lib.py +++ b/jc/lib.py @@ -3,35 +3,11 @@ import sys import os import re import importlib -from typing import Dict, List, Iterable, Union, Iterator +from typing import List, Iterable, Union, Iterator from types import ModuleType +from .jc_types import ParserInfoType, JSONDictType from jc import appdirs -if sys.version_info >= (3, 8): - from typing import TypedDict - ParserInfoType = TypedDict( - 'ParserInfoType', - { - "name": str, - "argument": str, - "version": str, - "description": str, - "author": str, - "author_email": str, - "compatible": List[str], - "magic_commands": List[str], - "documentation": str, - "streaming": bool, - "plugin": bool, - "hidden": bool, - "deprecated": bool - }, - total=False - ) -else: - ParserInfoType = Dict - -JSONDictType = Dict[str, Union[str, int, float, bool, List, Dict, None]] __version__ = '1.22.1' diff --git a/jc/parsers/foo.py b/jc/parsers/foo.py index 3b66980b..920aed95 100644 --- a/jc/parsers/foo.py +++ b/jc/parsers/foo.py @@ -33,7 +33,8 @@ Examples: $ foo | jc --foo -p -r [] """ -from typing import List, Dict +from typing import List +from jc.jc_types import JSONDictType import jc.utils @@ -53,7 +54,7 @@ class info(): __version__ = info.version -def _process(proc_data: List[Dict]) -> List[Dict]: +def _process(proc_data: List[JSONDictType]) -> List[JSONDictType]: """ Final processing to conform to the schema. @@ -78,7 +79,7 @@ def parse( data: str, raw: bool = False, quiet: bool = False -) -> List[Dict]: +) -> List[JSONDictType]: """ Main text parsing function @@ -95,7 +96,7 @@ def parse( jc.utils.compatibility(__name__, info.compatible, quiet) jc.utils.input_type_check(data) - raw_output: List = [] + raw_output: List[JSONDictType] = [] if jc.utils.has_data(data): diff --git a/jc/parsers/foo_s.py b/jc/parsers/foo_s.py index e1815a13..0c6a0e7d 100644 --- a/jc/parsers/foo_s.py +++ b/jc/parsers/foo_s.py @@ -45,6 +45,7 @@ import jc.utils from jc.streaming import ( add_jc_meta, streaming_input_type_check, streaming_line_input_type_check, raise_or_yield ) +from jc.jc_types import JSONDictType, StreamingOutputType from jc.exceptions import ParseError @@ -90,7 +91,7 @@ def parse( raw: bool = False, quiet: bool = False, ignore_exceptions: bool = False -) -> Union[Iterable[Dict], tuple]: +) -> StreamingOutputType: """ Main text parsing generator function. Returns an iterable object. @@ -114,7 +115,7 @@ def parse( for line in data: try: streaming_line_input_type_check(line) - output_line: Dict = {} + output_line: JSONDictType = {} # parse the content here # check out helper functions in jc.utils diff --git a/jc/streaming.py b/jc/streaming.py index 208e0687..995ac126 100644 --- a/jc/streaming.py +++ b/jc/streaming.py @@ -1,10 +1,11 @@ """jc - JSON Convert streaming utils""" from functools import wraps -from typing import Dict, Iterable +from typing import Union, Iterable +from .jc_types import JSONDictType, MetadataType -def streaming_input_type_check(data: Iterable) -> 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 `TypeError` if not. @@ -19,7 +20,7 @@ def streaming_line_input_type_check(line: str) -> None: raise TypeError("Input line must be a 'str' object.") -def stream_success(output_line: Dict, ignore_exceptions: bool) -> Dict: +def stream_success(output_line: JSONDictType, ignore_exceptions: bool) -> JSONDictType: """Add `_jc_meta` object to output line if `ignore_exceptions=True`""" if ignore_exceptions: output_line.update({'_jc_meta': {'success': True}}) @@ -27,7 +28,7 @@ def stream_success(output_line: Dict, ignore_exceptions: bool) -> Dict: return output_line -def stream_error(e: BaseException, line: str) -> Dict: +def stream_error(e: BaseException, line: str) -> MetadataType: """ Return an error `_jc_meta` field. """ @@ -103,7 +104,7 @@ def raise_or_yield( ignore_exceptions: bool, e: BaseException, line: str -) -> tuple: +) -> tuple[BaseException, str]: """ Return the exception object and line string if ignore_exceptions is True. Otherwise, re-raise the exception from the exception object with diff --git a/jc/utils.py b/jc/utils.py index 12e36bd8..e341faa6 100644 --- a/jc/utils.py +++ b/jc/utils.py @@ -7,19 +7,7 @@ from datetime import datetime, timezone from textwrap import TextWrapper from functools import lru_cache from typing import List, Dict, Iterable, Union, Optional, TextIO - -if sys.version_info >= (3, 8): - from typing import TypedDict - TSFormatType = TypedDict( - 'TSFormatType', - { - 'id': int, - 'format': str, - 'locale': Optional[str] - } - ) -else: - TSFormatType = Dict +from .jc_types import TimeStampFormatType def _asciify(string: str) -> str: @@ -394,7 +382,7 @@ class timestamp: If the conversion completely fails, all fields will be None. """ - formats: tuple[TSFormatType, ...] = ( + formats: tuple[TimeStampFormatType, ...] = ( {'id': 1000, 'format': '%a %b %d %H:%M:%S %Y', 'locale': None}, # manual C locale format conversion: Tue Mar 23 16:12:11 2021 or Tue Mar 23 16:12:11 IST 2021 {'id': 1100, 'format': '%a %b %d %H:%M:%S %Y %z', 'locale': None}, # git date output: Thu Mar 5 09:17:40 2020 -0800 {'id': 1300, 'format': '%Y-%m-%dT%H:%M:%S.%f%Z', 'locale': None}, # ISO Format with UTC (found in syslog 5424): 2003-10-11T22:14:15.003Z @@ -519,7 +507,7 @@ class timestamp: normalized_datetime = p.sub(r'\g<1> ', normalized_datetime) # try format hints first, then fall back to brute-force method - hint_obj_list: List[TSFormatType] = [] + hint_obj_list: List[TimeStampFormatType] = [] for fmt_id in format_hint: for fmt in formats: if fmt_id == fmt['id']: