Source code for ferrosoft.apps.dataimport.services.fixtures

import importlib
import json
from typing import Iterable, Tuple, Mapping, Callable

from jsonschema.validators import Draft202012Validator

from ferrosoft.apps.dataimport.services.errors import CatalogValidationError, CatalogConversionError

SchemaID = str
"""JSON schema ID"""




def load_converters(converters_name: str) -> Mapping[SchemaID, Callable]:
    module_name, member_name = converters_name.rsplit(".", 1)
    module = importlib.import_module(module_name)
    converters = getattr(module, member_name, None)
    if converters is None:
        raise RuntimeError("Converters not found: %s" % converters_name)
    if not isinstance(converters, dict):
        raise RuntimeError("Converters must be dict, got %s" % type(converters))
    return converters





def convert_catalog_files(
        catalog_file_names: Iterable[str],
        schema_file_name: str,
        converters: Mapping[SchemaID, Callable],
) -> list:
    """
    Create fixtures from catalog files.

    Args:
        catalog_file_names: JSON catalogs which must conform to JSON schema.
        schema_file_name: JSON schema file.
        converters: Converter function for each supported JSON schema: f(catalog) -> fixtures.

    Returns: Fixtures from all converted catalogs.
    """
    fixtures = []
    for catalog_path in catalog_file_names:
        with open(catalog_path, mode="rb") as catalog_file:
            catalog = json.load(catalog_file)
            schema_id, errors = validate_catalog(catalog, schema_file_name)
            if len(errors) > 0:
                raise CatalogValidationError(
                    "Invalid catalog file:\n" + "\n".join(errors)
                )
            if schema_id not in converters:
                raise CatalogConversionError("Cannot convert schema to fixtures: %s" % schema_id)
            fixtures.append(converters[schema_id](catalog))
    # Flatten fixtures.
    return [x for xs in fixtures for x in xs]





def validate_catalog(
        catalog: dict, schema_file_name: str
) -> Tuple[SchemaID, list]:
    """Validate catalog dict against catalog schema."""
    with open(schema_file_name, "rb") as schema_file:
        schema = json.load(schema_file)
        schema_id = schema.get("$id", None)
        validator = Draft202012Validator(schema)
        return SchemaID(schema_id), [
            "%s at %s" % (err.message, err.json_path)
            for err in validator.iter_errors(instance=catalog)
        ]