3822451468

Build Type

github

Committed by GitHub

Commit Message

Merge 8c3a0adb5 into cdd441319

Pull Request Pull Request #55: Add python3.11 to test matrix

Run Details

1553 of 1584 relevant lines covered (98.04%)

4.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.93

/json_to_models/cli.py

import argparse
import configparser
import importlib
import itertools
import json
import os.path
import re
import sys
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any, Callable, Dict, Generator, Iterable, List, Tuple, Type, Union

try:
    import ruamel.yaml as yaml
except ImportError:
    try:
        import yaml
    except ImportError:
        yaml = None

from . import __version__ as VERSION
from .dynamic_typing import ModelMeta, register_datetime_classes, registry
from .generator import MetadataGenerator
from .models import ModelsStructureType
from .models.attr import AttrsModelCodeGenerator
from .models.base import GenericModelCodeGenerator, generate_code
from .models.dataclasses import DataclassModelCodeGenerator
from .models.pydantic import PydanticModelCodeGenerator
from .models.structure import compose_models, compose_models_flat
from .registry import (
    ModelCmp, ModelFieldsEquals, ModelFieldsNumberMatch, ModelFieldsPercentMatch, ModelRegistry
)
from .utils import convert_args

STRUCTURE_FN_TYPE = Callable[[Dict[str, ModelMeta]], ModelsStructureType]
bool_js_style = lambda s: {"true": True, "false": False}.get(s, None)


class Cli:
    MODEL_CMP_MAPPING = {
        "percent": convert_args(ModelFieldsPercentMatch, lambda s: float(s) / 100),
        "number": convert_args(ModelFieldsNumberMatch, int),
        "exact": ModelFieldsEquals
    }

    STRUCTURE_FN_MAPPING: Dict[str, STRUCTURE_FN_TYPE] = {
        "nested": compose_models,
        "flat": compose_models_flat
    }

    MODEL_GENERATOR_MAPPING: Dict[str, Type[GenericModelCodeGenerator]] = {
        "base": convert_args(GenericModelCodeGenerator),
        "attrs": convert_args(AttrsModelCodeGenerator, meta=bool_js_style),
        "dataclasses": convert_args(DataclassModelCodeGenerator, meta=bool_js_style,
                                    post_init_converters=bool_js_style),
        "pydantic": convert_args(PydanticModelCodeGenerator),
    }

    def __init__(self):
        self.initialized = False
        self.models_data: Dict[str, Iterable[dict]] = {}  # -m/-l
        self.enable_datetime: bool = False  # --datetime
        self.strings_converters: bool = False  # --strings-converters
        self.max_literals: int = -1  # --max-strings-literals
        self.merge_policy: List[ModelCmp] = []  # --merge
        self.structure_fn: STRUCTURE_FN_TYPE = None  # -s
        self.model_generator: Type[GenericModelCodeGenerator] = None  # -f & --code-generator
        self.model_generator_kwargs: Dict[str, Any] = None

        self.argparser = self._create_argparser()

    def parse_args(self, args: List[str] = None):
        """
        Parse list of command list arguments

        :param args: (Optional) List of arguments
        :return: None
        """
        parser = self.argparser
        namespace = parser.parse_args(args)

        # Extract args
        parser = getattr(FileLoaders, namespace.input_format)
        self.output_file = namespace.output
        self.enable_datetime = namespace.datetime
        disable_unicode_conversion = namespace.disable_unicode_conversion
        self.strings_converters = namespace.strings_converters
        self.max_literals = namespace.max_strings_literals
        merge_policy = [m.split("_") if "_" in m else m for m in namespace.merge]
        structure = namespace.structure
        framework = namespace.framework
        code_generator = namespace.code_generator
        code_generator_kwargs_raw: List[str] = namespace.code_generator_kwargs
        dict_keys_regex: List[str] = namespace.dict_keys_regex
        dict_keys_fields: List[str] = namespace.dict_keys_fields
        preamble: str = namespace.preamble

        for name in namespace.disable_str_serializable_types:
            registry.remove_by_name(name)

        self.setup_models_data(namespace.model or (), namespace.list or (), parser)
        self.validate(merge_policy, framework, code_generator)
        self.set_args(merge_policy, structure, framework, code_generator, code_generator_kwargs_raw,
                      dict_keys_regex, dict_keys_fields, disable_unicode_conversion, preamble)

    def run(self):
        if self.enable_datetime:
            register_datetime_classes()
        generator = MetadataGenerator(
            dict_keys_regex=self.dict_keys_regex,
            dict_keys_fields=self.dict_keys_fields
        )
        registry = ModelRegistry(*self.merge_policy)
        for name, data in self.models_data.items():
            meta = generator.generate(*data)
            registry.process_meta_data(meta, name)
        registry.merge_models(generator)
        registry.generate_names()
        structure = self.structure_fn(registry.models_map)
        output = self.version_string + generate_code(
            structure,
            self.model_generator,
            class_generator_kwargs=self.model_generator_kwargs,
            preamble=self.preamble)
        if self.output_file:
            with open(self.output_file, "w", encoding="utf-8") as f:
                f.write(output)
            return f"Output is written to {self.output_file}"
        else:
            return output

    @property
    def version_string(self):
        return (
            'r"""\n'
            f'generated by json2python-models v{VERSION} at {datetime.now().ctime()}\n'
            f'command: {" ".join(sys.argv)}\n'
            '"""\n'
        )

    def validate(self, merge_policy, framework, code_generator):
        """
        Validate parsed args

        :param merge_policy: List of merge policies. Each merge policy is either string or string and policy arguments
        :param framework: Framework name (predefined code generator)
        :param code_generator: Code generator import string
        :return:
        """
        for m in merge_policy:
            if isinstance(m, list):
                if m[0] not in self.MODEL_CMP_MAPPING:
                    raise ValueError(f"Invalid merge policy '{m[0]}', choices are {self.MODEL_CMP_MAPPING.keys()}")
            elif m not in self.MODEL_CMP_MAPPING:
                raise ValueError(f"Invalid merge policy '{m}', choices are {self.MODEL_CMP_MAPPING.keys()}")

        if framework == 'custom' and code_generator is None:
            raise ValueError("You should specify --code-generator to support custom generator")
        elif framework != 'custom' and code_generator is not None:
            raise ValueError("--code-generator argument has no effect without '--framework custom' argument")

    def setup_models_data(
            self,
            models: Iterable[Union[
                Tuple[str, str],
                Tuple[str, str, str],
            ]],
            models_lists: Iterable[Tuple[str, str, str]],
            parser: 'FileLoaders.T'
    ):
        """
        Initialize lazy loaders for models data
        """
        models_dict: Dict[str, List[dict]] = defaultdict(list)

        models = list(models) + list(models_lists)
        for model_tuple in models:
            if len(model_tuple) == 2:
                model_name, path_raw = model_tuple
                lookup = '-'
            elif len(model_tuple) == 3:
                model_name, lookup, path_raw = model_tuple
            else:
                raise RuntimeError('`--model` argument should contain exactly 2 or 3 strings')

            for real_path in process_path(path_raw):
                iterator = iter_json_file(parser(real_path), lookup)
                models_dict[model_name].extend(iterator)

        self.models_data = models_dict

    def set_args(
            self,
            merge_policy: List[Union[List[str], str]],
            structure: str,
            framework: str,
            code_generator: str,
            code_generator_kwargs_raw: List[str],
            dict_keys_regex: List[str],
            dict_keys_fields: List[str],
            disable_unicode_conversion: bool,
            preamble: str,
    ):
        """
        Convert CLI args to python representation and set them to appropriate object attributes
        """
        self.merge_policy.clear()
        for merge in merge_policy:
            if isinstance(merge, str):
                name = merge
                args = ()
            else:
                name = merge[0]
                args = merge[1:]
            self.merge_policy.append(self.MODEL_CMP_MAPPING[name](*args))

        self.structure_fn = self.STRUCTURE_FN_MAPPING[structure]

        if framework != "custom":
            self.model_generator = self.MODEL_GENERATOR_MAPPING[framework]
        else:
            module, cls = code_generator.rsplit('.', 1)
            m = importlib.import_module(module)
            self.model_generator = getattr(m, cls)

        self.model_generator_kwargs = dict(
            post_init_converters=self.strings_converters,
            convert_unicode=not disable_unicode_conversion,
            max_literals=self.max_literals
        )
        if code_generator_kwargs_raw:
            for item in code_generator_kwargs_raw:
                if item[0] == '"':
                    item = item[1:]
                if item[-1] == '"':
                    item = item[:-1]
                name, value = item.split("=", 1)
                self.model_generator_kwargs[name] = value

        self.dict_keys_regex = [re.compile(rf"^{r}$") for r in dict_keys_regex] if dict_keys_regex else ()
        self.dict_keys_fields = dict_keys_fields or ()
        if preamble:
            preamble = preamble.strip()
        self.preamble = preamble or None
        self.initialized = True

    @classmethod
    def _create_argparser(cls) -> argparse.ArgumentParser:
        """
        ArgParser factory
        """
        parser = argparse.ArgumentParser(
            formatter_class=argparse.RawTextHelpFormatter,
            description="Convert given json files into Python models."
        )

        parser.add_argument(
            "-m", "--model",
            nargs="+", action="append", metavar=("<Model name> [<JSON lookup>] <File path or pattern>", ""),
            help="Model name and its JSON data as path or unix-like path pattern.\n"
                 "'*',  '**' or '?' patterns symbols are supported.\n\n"
                 "JSON data could be array of models or single model\n\n"
                 "If this file contains dict with nested list than you can pass\n"
                 "<JSON lookup>. Deep lookups are supported by dot-separated path.\n"
                 "If no lookup needed pass '-' as <JSON lookup> (default)\n\n"
        )
        parser.add_argument(
            "-i", "--input-format",
            default="json",
            choices=['json', 'yaml', 'ini'],
            help="Input files parser ('PyYaml' is required to parse yaml files)\n\n"
        )
        parser.add_argument(
            "-o", "--output",
            metavar="FILE", default="",
            help="Path to output file\n\n"
        )
        parser.add_argument(
            "-f", "--framework",
            default="base",
            choices=list(cls.MODEL_GENERATOR_MAPPING.keys()) + ["custom"],
            help="Model framework for which python code is generated.\n"
                 "'base' (default) mean no framework so code will be generated without any decorators\n"
                 "and additional meta-data.\n"
                 "If you pass 'custom' you should specify --code-generator argument\n\n"
        )
        parser.add_argument(
            "-s", "--structure",
            default="flat",
            choices=list(cls.STRUCTURE_FN_MAPPING.keys()),
            help="Models composition style. By default nested models become nested Python classes.\n\n"
        )
        parser.add_argument(
            "--datetime",
            action="store_true",
            help="Enable datetime/date/time strings parsing.\n"
                 "Warn.: This can lead to 6-7 times slowdown on large datasets.\n"
                 "       Be sure that you really need this option.\n\n"
        )
        parser.add_argument(
            "--strings-converters",
            action="store_true",
            help="Enable generation of string types converters (i.e. IsoDatetimeString or BooleanString).\n\n"
        )
        parser.add_argument(
            "--max-strings-literals",
            type=int,
            default=GenericModelCodeGenerator.DEFAULT_MAX_LITERALS,
            metavar='NUMBER',
            help="Generate Literal['foo', 'bar'] when field have less than NUMBER string constants as values.\n"
                 f"Pass 0 to disable. By default NUMBER={GenericModelCodeGenerator.DEFAULT_MAX_LITERALS}"
                 f" (some generator classes could override it)\n\n"
        )
        parser.add_argument(
            "--disable-unicode-conversion", "--no-unidecode",
            action="store_true",
            help="Disabling unicode conversion in fields and class names.\n\n"
        )

        default_percent = f"{ModelFieldsPercentMatch.DEFAULT * 100:.0f}"
        default_number = f"{ModelFieldsNumberMatch.DEFAULT:.0f}"
        parser.add_argument(
            "--merge",
            default=["percent", "number"],
            nargs="+",
            help=(
                f"Merge policy settings. Default is 'percent_{default_percent} number_{default_number}' (percent of field match\n"
                "or number of fields match).\n"
                "Possible values are:\n"
                "'percent[_<percent>]' - two models had a certain percentage of matched field names.\n"
                f"                        Default percent is {default_percent}%%. "
                "Custom value could be i.e. 'percent_95'.\n"
                "'number[_<number>]'   - two models had a certain number of matched field names.\n"
                f"                        Default number of fields is {default_number}.\n"
                "'exact'               - two models should have exact same field names to merge.\n\n"
            )
        )
        parser.add_argument(
            "--dict-keys-regex", "--dkr",
            nargs="+", metavar="RegEx",
            help="List of regular expressions (Python syntax).\n"
                 "If all keys of some dict are match one of them\n"
                 "then this dict will be marked as dict field but not nested model.\n"
                 "Note: ^ and $ tokens will be added automatically but you have to\n"
                 "escape other special characters manually.\n"
        )
        parser.add_argument(
            "--dict-keys-fields", "--dkf",
            nargs="+", metavar="FIELD NAME",
            help="List of model fields names that will be marked as dict fields\n\n"
        )
        parser.add_argument(
            "--code-generator",
            help="Absolute import path to GenericModelCodeGenerator subclass.\n"
                 "Works in pair with '-f custom'\n\n"
        )
        parser.add_argument(
            "--code-generator-kwargs",
            metavar="NAME=VALUE",
            nargs="*", type=str,
            help="List of code generator arguments (for __init__ method).\n"
                 "Each argument should be in following format:\n"
                 "    argument_name=value or \"argument_name=value with space\"\n"
                 "Boolean values should be passed in JS style: true | false"
                 "\n\n"
        )
        parser.add_argument(
            "--preamble",
            type=str,
            help="Code to insert into the generated file after the imports and before the list of classes\n\n"
        )
        parser.add_argument(
            "--disable-str-serializable-types",
            metavar="TYPE",
            default=[],
            nargs="*", type=str,
            help="List of python types for which StringSerializable should be disabled, i.e:\n"
                 "--disable-str-serializable-types float int\n"
                 "Alternatively you could use the name of StringSerializable subclass itself (i.e. IntString)"
                 "\n\n"
        )
        parser.add_argument(
            "-l", "--list",
            nargs=3, action="append", metavar=("<Model name>", "<JSON lookup>", "<JSON file>"),
            help="DEPRECATED, use --model argument instead"
        )

        return parser


def main():
    import os

    if os.getenv("TRAVIS", None) or os.getenv("FORCE_COVERAGE", None):
        # Enable coverage if it is Travis-CI or env variable FORCE_COVERAGE set to true
        import coverage

        coverage.process_startup()

    cli = Cli()
    cli.parse_args()
    print(cli.run())


class FileLoaders:
    T = Callable[[Path], Union[dict, list]]

    @staticmethod
    def json(path: Path) -> Union[dict, list]:
        with path.open() as fp:
            return json.load(fp)

    @staticmethod
    def yaml(path: Path) -> Union[dict, list]:
        if yaml is None:
            print('Yaml parser is not installed. To parse yaml files PyYaml (or ruamel.yaml) is required.')
            raise ImportError('yaml')
        with path.open() as fp:
            return yaml.safe_load(fp)

    @staticmethod
    def ini(path: Path) -> dict:
        config = configparser.ConfigParser()
        with path.open() as fp:
            config.read_file(fp)
        return {s: dict(config.items(s)) for s in config.sections()}


def dict_lookup(d: Union[dict, list], lookup: str) -> Union[dict, list]:
    """
    Extract nested value from key path.
    If lookup is "-" returns dict as is.

    :param d: Nested dict
    :param lookup: Dot separated lookup path
    :return: Nested value
    """
    while lookup and lookup != "-":
        split = lookup.split('.', 1)
        if len(split) == 1:
            return d[split[0]]
        key, lookup = split
        d = d[key]
    return d


def iter_json_file(data: Union[dict, list], lookup: str) -> Generator[Union[dict, list], Any, None]:
    """
    Perform lookup and return generator over json list.
    Does not open file until iteration is started.

    :param data: JSON data
    :param lookup: Dot separated lookup path
    :return: Generator of the model data
    """
    item = dict_lookup(data, lookup)
    if isinstance(item, list):
        yield from item
    elif isinstance(item, dict):
        yield item
    else:
        raise TypeError(f'dict or list is expected at {lookup if lookup != "-" else "JSON root"}, not {type(item)}')


def process_path(path: str) -> Iterable[Path]:
    """
    Convert path pattern into path iterable.
    If non-pattern path is given return tuple of one element: (path,)
    """
    split_path = path_split(path)
    clean_path = list(itertools.takewhile(
        lambda part: "*" not in part and "?" not in part,
        split_path
    ))
    pattern_path = split_path[len(clean_path):]

    if clean_path:
        clean_path = os.path.join(*clean_path)
    else:
        clean_path = "."

    if pattern_path:
        pattern_path = os.path.join(*pattern_path)
    else:
        pattern_path = None

    path = Path(clean_path)
    if pattern_path:
        return path.glob(pattern_path)
    else:
        return path,


def path_split(path: str) -> List[str]:
    """
    Split path into list of components

    :param path: string path
    :return: List of files/patterns
    """
    folders = []
    while True:
        path, folder = os.path.split(path)

        if folder:
            folders.append(folder)
        else:
            if path:
                folders.append(path)
            break
    folders.reverse()
    return folders

1	import argparse	5✔
2	import configparser	5✔
3	import importlib	5✔
4	import itertools	5✔
5	import json	5✔
6	import os.path	5✔
7	import re	5✔
8	import sys	5✔
9	from collections import defaultdict	5✔
10	from datetime import datetime	5✔
11	from pathlib import Path	5✔
12	from typing import Any, Callable, Dict, Generator, Iterable, List, Tuple, Type, Union	5✔
13
14	try:	5✔
15	import ruamel.yaml as yaml	5✔
16	except ImportError:	×
17	try:	×
18	import yaml	×
19	except ImportError:	×
20	yaml = None	×
21
22	from . import __version__ as VERSION	5✔
23	from .dynamic_typing import ModelMeta, register_datetime_classes, registry	5✔
24	from .generator import MetadataGenerator	5✔
25	from .models import ModelsStructureType	5✔
26	from .models.attr import AttrsModelCodeGenerator	5✔
27	from .models.base import GenericModelCodeGenerator, generate_code	5✔
28	from .models.dataclasses import DataclassModelCodeGenerator	5✔
29	from .models.pydantic import PydanticModelCodeGenerator	5✔
30	from .models.structure import compose_models, compose_models_flat	5✔
31	from .registry import (	5✔
32	ModelCmp, ModelFieldsEquals, ModelFieldsNumberMatch, ModelFieldsPercentMatch, ModelRegistry
33	)
34	from .utils import convert_args	5✔
35
36	STRUCTURE_FN_TYPE = Callable[[Dict[str, ModelMeta]], ModelsStructureType]	5✔
37	bool_js_style = lambda s: {"true": True, "false": False}.get(s, None)	5✔
38
39
40	class Cli:	5✔
41	MODEL_CMP_MAPPING = {	5✔
42	"percent": convert_args(ModelFieldsPercentMatch, lambda s: float(s) / 100),
43	"number": convert_args(ModelFieldsNumberMatch, int),
44	"exact": ModelFieldsEquals
45	}
46
47	STRUCTURE_FN_MAPPING: Dict[str, STRUCTURE_FN_TYPE] = {	5✔
48	"nested": compose_models,
49	"flat": compose_models_flat
50	}
51
52	MODEL_GENERATOR_MAPPING: Dict[str, Type[GenericModelCodeGenerator]] = {	5✔
53	"base": convert_args(GenericModelCodeGenerator),
54	"attrs": convert_args(AttrsModelCodeGenerator, meta=bool_js_style),
55	"dataclasses": convert_args(DataclassModelCodeGenerator, meta=bool_js_style,
56	post_init_converters=bool_js_style),
57	"pydantic": convert_args(PydanticModelCodeGenerator),
58	}
59
60	def __init__(self):	5✔
61	self.initialized = False	5✔
62	self.models_data: Dict[str, Iterable[dict]] = {} # -m/-l	5✔
63	self.enable_datetime: bool = False # --datetime	5✔
64	self.strings_converters: bool = False # --strings-converters	5✔
65	self.max_literals: int = -1 # --max-strings-literals	5✔
66	self.merge_policy: List[ModelCmp] = [] # --merge	5✔
67	self.structure_fn: STRUCTURE_FN_TYPE = None # -s	5✔
68	self.model_generator: Type[GenericModelCodeGenerator] = None # -f & --code-generator	5✔
69	self.model_generator_kwargs: Dict[str, Any] = None	5✔
70
71	self.argparser = self._create_argparser()	5✔
72
73	def parse_args(self, args: List[str] = None):	5✔
74	"""
75	Parse list of command list arguments
76
77	:param args: (Optional) List of arguments
78	:return: None
79	"""
80	parser = self.argparser	5✔
81	namespace = parser.parse_args(args)	5✔
82
83	# Extract args
84	parser = getattr(FileLoaders, namespace.input_format)	5✔
85	self.output_file = namespace.output	5✔
86	self.enable_datetime = namespace.datetime	5✔
87	disable_unicode_conversion = namespace.disable_unicode_conversion	5✔
88	self.strings_converters = namespace.strings_converters	5✔
89	self.max_literals = namespace.max_strings_literals	5✔
90	merge_policy = [m.split("_") if "_" in m else m for m in namespace.merge]	5✔
91	structure = namespace.structure	5✔
92	framework = namespace.framework	5✔
93	code_generator = namespace.code_generator	5✔
94	code_generator_kwargs_raw: List[str] = namespace.code_generator_kwargs	5✔
95	dict_keys_regex: List[str] = namespace.dict_keys_regex	5✔
96	dict_keys_fields: List[str] = namespace.dict_keys_fields	5✔
97	preamble: str = namespace.preamble	5✔
98
99	for name in namespace.disable_str_serializable_types:	5✔
100	registry.remove_by_name(name)	5✔
101
102	self.setup_models_data(namespace.model or (), namespace.list or (), parser)	5✔
103	self.validate(merge_policy, framework, code_generator)	5✔
104	self.set_args(merge_policy, structure, framework, code_generator, code_generator_kwargs_raw,	5✔
105	dict_keys_regex, dict_keys_fields, disable_unicode_conversion, preamble)
106
107	def run(self):	5✔
108	if self.enable_datetime:	5✔
109	register_datetime_classes()	5✔
110	generator = MetadataGenerator(	5✔
111	dict_keys_regex=self.dict_keys_regex,
112	dict_keys_fields=self.dict_keys_fields
113	)
114	registry = ModelRegistry(*self.merge_policy)	5✔
115	for name, data in self.models_data.items():	5✔
116	meta = generator.generate(*data)	5✔
117	registry.process_meta_data(meta, name)	5✔
118	registry.merge_models(generator)	5✔
119	registry.generate_names()	5✔
120	structure = self.structure_fn(registry.models_map)	5✔
121	output = self.version_string + generate_code(	5✔
122	structure,
123	self.model_generator,
124	class_generator_kwargs=self.model_generator_kwargs,
125	preamble=self.preamble)
126	if self.output_file:	5✔
127	with open(self.output_file, "w", encoding="utf-8") as f:	5✔
128	f.write(output)	5✔
129	return f"Output is written to {self.output_file}"	5✔
130	else:
131	return output	5✔
132
133	@property	5✔
134	def version_string(self):	4✔
135	return (	5✔
136	'r"""\n'
137	f'generated by json2python-models v{VERSION} at {datetime.now().ctime()}\n'
138	f'command: {" ".join(sys.argv)}\n'
139	'"""\n'
140	)
141
142	def validate(self, merge_policy, framework, code_generator):	5✔
143	"""
144	Validate parsed args
145
146	:param merge_policy: List of merge policies. Each merge policy is either string or string and policy arguments
147	:param framework: Framework name (predefined code generator)
148	:param code_generator: Code generator import string
149	:return:
150	"""
151	for m in merge_policy:	5✔
152	if isinstance(m, list):	5✔
153	if m[0] not in self.MODEL_CMP_MAPPING:	5✔
154	raise ValueError(f"Invalid merge policy '{m[0]}', choices are {self.MODEL_CMP_MAPPING.keys()}")	5✔
155	elif m not in self.MODEL_CMP_MAPPING:	5✔
156	raise ValueError(f"Invalid merge policy '{m}', choices are {self.MODEL_CMP_MAPPING.keys()}")	5✔
157
158	if framework == 'custom' and code_generator is None:	5✔
159	raise ValueError("You should specify --code-generator to support custom generator")	5✔
160	elif framework != 'custom' and code_generator is not None:	5✔
161	raise ValueError("--code-generator argument has no effect without '--framework custom' argument")	5✔
162
163	def setup_models_data(	5✔
164	self,
165	models: Iterable[Union[
166	Tuple[str, str],
167	Tuple[str, str, str],
168	]],
169	models_lists: Iterable[Tuple[str, str, str]],
170	parser: 'FileLoaders.T'
171	):
172	"""
173	Initialize lazy loaders for models data
174	"""
175	models_dict: Dict[str, List[dict]] = defaultdict(list)	5✔
176
177	models = list(models) + list(models_lists)	5✔
178	for model_tuple in models:	5✔
179	if len(model_tuple) == 2:	5✔
180	model_name, path_raw = model_tuple	5✔
181	lookup = '-'	5✔
182	elif len(model_tuple) == 3:	5✔
183	model_name, lookup, path_raw = model_tuple	5✔
184	else:
185	raise RuntimeError('`--model` argument should contain exactly 2 or 3 strings')	5✔
186
187	for real_path in process_path(path_raw):	5✔
188	iterator = iter_json_file(parser(real_path), lookup)	5✔
189	models_dict[model_name].extend(iterator)	5✔
190
191	self.models_data = models_dict	5✔
192
193	def set_args(	5✔
194	self,
195	merge_policy: List[Union[List[str], str]],
196	structure: str,
197	framework: str,
198	code_generator: str,
199	code_generator_kwargs_raw: List[str],
200	dict_keys_regex: List[str],
201	dict_keys_fields: List[str],
202	disable_unicode_conversion: bool,
203	preamble: str,
204	):
205	"""
206	Convert CLI args to python representation and set them to appropriate object attributes
207	"""
208	self.merge_policy.clear()	5✔
209	for merge in merge_policy:	5✔
210	if isinstance(merge, str):	5✔
211	name = merge	5✔
212	args = ()	5✔
213	else:
214	name = merge[0]	5✔
215	args = merge[1:]	5✔
216	self.merge_policy.append(self.MODEL_CMP_MAPPING[name](*args))	5✔
217
218	self.structure_fn = self.STRUCTURE_FN_MAPPING[structure]	5✔
219
220	if framework != "custom":	5✔
221	self.model_generator = self.MODEL_GENERATOR_MAPPING[framework]	5✔
222	else:
223	module, cls = code_generator.rsplit('.', 1)	5✔
224	m = importlib.import_module(module)	5✔
225	self.model_generator = getattr(m, cls)	5✔
226
227	self.model_generator_kwargs = dict(	5✔
228	post_init_converters=self.strings_converters,
229	convert_unicode=not disable_unicode_conversion,
230	max_literals=self.max_literals
231	)
232	if code_generator_kwargs_raw:	5✔
233	for item in code_generator_kwargs_raw:	5✔
234	if item[0] == '"':	5✔
235	item = item[1:]	×
236	if item[-1] == '"':	5✔
237	item = item[:-1]	×
238	name, value = item.split("=", 1)	5✔
239	self.model_generator_kwargs[name] = value	5✔
240
241	self.dict_keys_regex = [re.compile(rf"^{r}$") for r in dict_keys_regex] if dict_keys_regex else ()	5✔
242	self.dict_keys_fields = dict_keys_fields or ()	5✔
243	if preamble:	5✔
244	preamble = preamble.strip()	5✔
245	self.preamble = preamble or None	5✔
246	self.initialized = True	5✔
247
248	@classmethod	5✔
249	def _create_argparser(cls) -> argparse.ArgumentParser:	5✔
250	"""
251	ArgParser factory
252	"""
253	parser = argparse.ArgumentParser(	5✔
254	formatter_class=argparse.RawTextHelpFormatter,
255	description="Convert given json files into Python models."
256	)
257
258	parser.add_argument(
259	"-m", "--model",
260	nargs="+", action="append", metavar=("<Model name> [<JSON lookup>] <File path or pattern>", ""),
261	help="Model name and its JSON data as path or unix-like path pattern.\n"
262	"'', '*' or '?' patterns symbols are supported.\n\n"
263	"JSON data could be array of models or single model\n\n"
264	"If this file contains dict with nested list than you can pass\n"
265	"<JSON lookup>. Deep lookups are supported by dot-separated path.\n"
266	"If no lookup needed pass '-' as <JSON lookup> (default)\n\n"
267	)
268	parser.add_argument(	5✔
269	"-i", "--input-format",
270	default="json",
271	choices=['json', 'yaml', 'ini'],
272	help="Input files parser ('PyYaml' is required to parse yaml files)\n\n"
273	)
274	parser.add_argument(	5✔
275	"-o", "--output",
276	metavar="FILE", default="",
277	help="Path to output file\n\n"
278	)
279	parser.add_argument(
280	"-f", "--framework",
281	default="base",
282	choices=list(cls.MODEL_GENERATOR_MAPPING.keys()) + ["custom"],
283	help="Model framework for which python code is generated.\n"
284	"'base' (default) mean no framework so code will be generated without any decorators\n"
285	"and additional meta-data.\n"
286	"If you pass 'custom' you should specify --code-generator argument\n\n"
287	)
288	parser.add_argument(	5✔
289	"-s", "--structure",
290	default="flat",
291	choices=list(cls.STRUCTURE_FN_MAPPING.keys()),
292	help="Models composition style. By default nested models become nested Python classes.\n\n"
293	)
294	parser.add_argument(	5✔
295	"--datetime",
296	action="store_true",
297	help="Enable datetime/date/time strings parsing.\n"
298	"Warn.: This can lead to 6-7 times slowdown on large datasets.\n"
299	" Be sure that you really need this option.\n\n"
300	)
301	parser.add_argument(	5✔
302	"--strings-converters",
303	action="store_true",
304	help="Enable generation of string types converters (i.e. IsoDatetimeString or BooleanString).\n\n"
305	)
306	parser.add_argument(	5✔
307	"--max-strings-literals",
308	type=int,
309	default=GenericModelCodeGenerator.DEFAULT_MAX_LITERALS,
310	metavar='NUMBER',
311	help="Generate Literal['foo', 'bar'] when field have less than NUMBER string constants as values.\n"
312	f"Pass 0 to disable. By default NUMBER={GenericModelCodeGenerator.DEFAULT_MAX_LITERALS}"
313	f" (some generator classes could override it)\n\n"
314	)
315	parser.add_argument(	5✔
316	"--disable-unicode-conversion", "--no-unidecode",
317	action="store_true",
318	help="Disabling unicode conversion in fields and class names.\n\n"
319	)
320
321	default_percent = f"{ModelFieldsPercentMatch.DEFAULT * 100:.0f}"	5✔
322	default_number = f"{ModelFieldsNumberMatch.DEFAULT:.0f}"	5✔
323	parser.add_argument(	5✔
324	"--merge",
325	default=["percent", "number"],
326	nargs="+",
327	help=(
328	f"Merge policy settings. Default is 'percent_{default_percent} number_{default_number}' (percent of field match\n"
329	"or number of fields match).\n"
330	"Possible values are:\n"
331	"'percent[_<percent>]' - two models had a certain percentage of matched field names.\n"
332	f" Default percent is {default_percent}%%. "
333	"Custom value could be i.e. 'percent_95'.\n"
334	"'number[_<number>]' - two models had a certain number of matched field names.\n"
335	f" Default number of fields is {default_number}.\n"
336	"'exact' - two models should have exact same field names to merge.\n\n"
337	)
338	)
339	parser.add_argument(	5✔
340	"--dict-keys-regex", "--dkr",
341	nargs="+", metavar="RegEx",
342	help="List of regular expressions (Python syntax).\n"
343	"If all keys of some dict are match one of them\n"
344	"then this dict will be marked as dict field but not nested model.\n"
345	"Note: ^ and $ tokens will be added automatically but you have to\n"
346	"escape other special characters manually.\n"
347	)
348	parser.add_argument(	5✔
349	"--dict-keys-fields", "--dkf",
350	nargs="+", metavar="FIELD NAME",
351	help="List of model fields names that will be marked as dict fields\n\n"
352	)
353	parser.add_argument(	5✔
354	"--code-generator",
355	help="Absolute import path to GenericModelCodeGenerator subclass.\n"
356	"Works in pair with '-f custom'\n\n"
357	)
358	parser.add_argument(
359	"--code-generator-kwargs",
360	metavar="NAME=VALUE",
361	nargs="*", type=str,
362	help="List of code generator arguments (for __init__ method).\n"
363	"Each argument should be in following format:\n"
364	" argument_name=value or \"argument_name=value with space\"\n"
365	"Boolean values should be passed in JS style: true \| false"
366	"\n\n"
367	)
368	parser.add_argument(	5✔
369	"--preamble",
370	type=str,
371	help="Code to insert into the generated file after the imports and before the list of classes\n\n"
372	)
373	parser.add_argument(	5✔
374	"--disable-str-serializable-types",
375	metavar="TYPE",
376	default=[],
377	nargs="*", type=str,
378	help="List of python types for which StringSerializable should be disabled, i.e:\n"
379	"--disable-str-serializable-types float int\n"
380	"Alternatively you could use the name of StringSerializable subclass itself (i.e. IntString)"
381	"\n\n"
382	)
383	parser.add_argument(	5✔
384	"-l", "--list",
385	nargs=3, action="append", metavar=("<Model name>", "<JSON lookup>", "<JSON file>"),
386	help="DEPRECATED, use --model argument instead"
387	)
388
389	return parser	5✔
390
391
392	def main():
393	import os
394
395	if os.getenv("TRAVIS", None) or os.getenv("FORCE_COVERAGE", None):
396	# Enable coverage if it is Travis-CI or env variable FORCE_COVERAGE set to true
397	import coverage
398
399	coverage.process_startup()
400
401	cli = Cli()
402	cli.parse_args()
403	print(cli.run())
404
405
406	class FileLoaders:	5✔
407	T = Callable[[Path], Union[dict, list]]	5✔
408
409	@staticmethod	5✔
410	def json(path: Path) -> Union[dict, list]:	5✔
411	with path.open() as fp:	5✔
412	return json.load(fp)	5✔
413
414	@staticmethod	5✔
415	def yaml(path: Path) -> Union[dict, list]:	5✔
416	if yaml is None:	5✔
417	print('Yaml parser is not installed. To parse yaml files PyYaml (or ruamel.yaml) is required.')	×
418	raise ImportError('yaml')	×
419	with path.open() as fp:	5✔
420	return yaml.safe_load(fp)	5✔
421
422	@staticmethod	5✔
423	def ini(path: Path) -> dict:	5✔
424	config = configparser.ConfigParser()	5✔
425	with path.open() as fp:	5✔
426	config.read_file(fp)	5✔
427	return {s: dict(config.items(s)) for s in config.sections()}	5✔
428
429
430	def dict_lookup(d: Union[dict, list], lookup: str) -> Union[dict, list]:	5✔
431	"""
432	Extract nested value from key path.
433	If lookup is "-" returns dict as is.
434
435	:param d: Nested dict
436	:param lookup: Dot separated lookup path
437	:return: Nested value
438	"""
439	while lookup and lookup != "-":	5✔
440	split = lookup.split('.', 1)	5✔
441	if len(split) == 1:	5✔
442	return d[split[0]]	5✔
443	key, lookup = split	5✔
444	d = d[key]	5✔
445	return d	5✔
446
447
448	def iter_json_file(data: Union[dict, list], lookup: str) -> Generator[Union[dict, list], Any, None]:	5✔
449	"""
450	Perform lookup and return generator over json list.
451	Does not open file until iteration is started.
452
453	:param data: JSON data
454	:param lookup: Dot separated lookup path
455	:return: Generator of the model data
456	"""
457	item = dict_lookup(data, lookup)	5✔
458	if isinstance(item, list):	5✔
459	yield from item	5✔
460	elif isinstance(item, dict):	5✔
461	yield item	5✔
462	else:
463	raise TypeError(f'dict or list is expected at {lookup if lookup != "-" else "JSON root"}, not {type(item)}')	5✔
464
465
466	def process_path(path: str) -> Iterable[Path]:	5✔
467	"""
468	Convert path pattern into path iterable.
469	If non-pattern path is given return tuple of one element: (path,)
470	"""
471	split_path = path_split(path)	5✔
472	clean_path = list(itertools.takewhile(	5✔
473	lambda part: "*" not in part and "?" not in part,
474	split_path
475	))
476	pattern_path = split_path[len(clean_path):]	5✔
477
478	if clean_path:	5✔
479	clean_path = os.path.join(*clean_path)	5✔
480	else:
481	clean_path = "."	5✔
482
483	if pattern_path:	5✔
484	pattern_path = os.path.join(*pattern_path)	5✔
485	else:
486	pattern_path = None	5✔
487
488	path = Path(clean_path)	5✔
489	if pattern_path:	5✔
490	return path.glob(pattern_path)	5✔
491	else:
492	return path,	5✔
493
494
495	def path_split(path: str) -> List[str]:	5✔
496	"""
497	Split path into list of components
498
499	:param path: string path
500	:return: List of files/patterns
501	"""
502	folders = []	5✔
503	while True:	3✔
504	path, folder = os.path.split(path)	5✔
505
506	if folder:	5✔
507	folders.append(folder)	5✔
508	else:
509	if path:	5✔
510	folders.append(path)	5✔
511	break	5✔
512	folders.reverse()	5✔
513	return folders	5✔

bogdandm / json2python-models / 3822451468

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous