1
0
mirror of https://github.com/kellyjonbrazil/jc.git synced 2025-06-23 00:29:59 +02:00

move custom types to jc_types

This commit is contained in:
Kelly Brazil
2022-10-18 11:01:59 -07:00
parent 3639e02843
commit dfa7a71f53
7 changed files with 78 additions and 60 deletions

View File

@ -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

55
jc/jc_types.py Normal file
View File

@ -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

View File

@ -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'

View File

@ -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):

View File

@ -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

View File

@ -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

View File

@ -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']: