10202339569

Committed 01 Aug 2024 04:44PM UTC coverage: 95.709%. Remained the same

Build # 10202339569

Build Type

push

github

Committed by

Bjwebb

Commit Message

errors: Use custom exceptions

https://github.com/OpenDataServices/flatten-tool/issues/450

This makes it easier to diambiguate errors deliberately raised by
flatten-tool versus those from other sources. I've left alone a few
exceptions that flatten-tool raises, but which we don't expect to
happen, so didn't seem to be in the same category.

Run Details

20 of 35 new or added lines in 6 files covered. (57.14%)

57 existing lines in 5 files now uncovered.

3390 of 3542 relevant lines covered (95.71%)

11.42 hits per line

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

94.43

/flattentool/json_input.py

"""

This file contains code that takes an instance of a JSON file as input (not a
JSON schema, for that see schema.py).

"""

import codecs
import copy
import os
import tempfile
import uuid
from collections import OrderedDict
from decimal import Decimal
from warnings import warn

import BTrees.OOBTree
import ijson

try:
    import shapely.wkt

    SHAPELY_LIBRARY_AVAILABLE = True
except ImportError:
    SHAPELY_LIBRARY_AVAILABLE = False
import transaction
import xmltodict
import zc.zlibstorage
import ZODB.FileStorage

from flattentool.exceptions import (
    DataErrorWarning,
    FlattenToolError,
    FlattenToolValueError,
    FlattenToolWarning,
)
from flattentool.i18n import _
from flattentool.input import path_search
from flattentool.schema import make_sub_sheet_name
from flattentool.sheet import PersistentSheet

BASIC_TYPES = [str, bool, int, Decimal, type(None)]


class BadlyFormedJSONError(FlattenToolError, ValueError):
    pass


class BadlyFormedJSONErrorUTF8(BadlyFormedJSONError):
    pass


def sheet_key_field(sheet, key):
    if key not in sheet:
        sheet.append(key)
    return key


def sheet_key_title(sheet, key):
    """
    If the key has a corresponding title, return that. If doesn't, create it in the sheet and return it.

    """
    if key in sheet.titles:
        title = sheet.titles[key]
        if title not in sheet:
            sheet.append(title)
        return title
    else:
        if key not in sheet:
            sheet.append(key)
        return key


def lists_of_dicts_paths(xml_dict):
    for key, value in xml_dict.items():
        if isinstance(value, list) and value and isinstance(value[0], dict):
            yield (key,)
            for x in value:
                if isinstance(x, dict):
                    for path in lists_of_dicts_paths(x):
                        yield (key,) + path
        elif isinstance(value, dict):
            for path in lists_of_dicts_paths(value):
                yield (key,) + path


def dicts_to_list_of_dicts(lists_of_dicts_paths_set, xml_dict, path=()):
    for key, value in xml_dict.items():
        if isinstance(value, list):
            for x in value:
                if isinstance(x, dict):
                    dicts_to_list_of_dicts(lists_of_dicts_paths_set, x, path + (key,))
        elif isinstance(value, dict):
            child_path = path + (key,)
            dicts_to_list_of_dicts(lists_of_dicts_paths_set, value, child_path)
            if child_path in lists_of_dicts_paths_set:
                xml_dict[key] = [value]


def list_dict_consistency(xml_dict):
    """
    For use with XML files opened with xmltodict.

    If there is only one tag, xmltodict produces a dict. If there are
    multiple, xmltodict produces a list of dicts. This functions replaces
    dicts with lists of dicts, if there exists a list of dicts for the same
    path elsewhere in the file.
    """
    lists_of_dicts_paths_set = set(lists_of_dicts_paths(xml_dict))
    dicts_to_list_of_dicts(lists_of_dicts_paths_set, xml_dict)


class JSONParser(object):
    # Named for consistency with schema.SchemaParser, but not sure it's the most appropriate name.
    # Similarly with methods like parse_json_dict

    def __init__(
        self,
        json_filename=None,
        root_json_dict=None,
        schema_parser=None,
        root_list_path=None,
        root_id="ocid",
        use_titles=False,
        xml=False,
        id_name="id",
        filter_field=None,
        filter_value=None,
        preserve_fields=None,
        remove_empty_schema_columns=False,
        rollup=False,
        truncation_length=3,
        persist=False,
        convert_flags={},
    ):
        if persist:
            # Use temp directories in OS agnostic way
            self.zodb_db_location = (
                tempfile.gettempdir() + "/flattentool-" + str(uuid.uuid4())
            )
            # zlibstorage lowers disk usage by a lot at very small performance cost
            zodb_storage = zc.zlibstorage.ZlibStorage(
                ZODB.FileStorage.FileStorage(self.zodb_db_location)
            )
            self.db = ZODB.DB(zodb_storage)
        else:
            # If None, in memory storage is used.
            self.db = ZODB.DB(None)

        self.connection = self.db.open()

        # ZODB root, only objects attached here will be persisted
        root = self.connection.root
        # OOBTree means a btree with keys and values are objects (including strings)
        root.sheet_store = BTrees.OOBTree.BTree()

        self.sub_sheets = {}
        self.main_sheet = PersistentSheet(connection=self.connection, name="")
        self.root_list_path = root_list_path
        self.root_id = root_id
        self.use_titles = use_titles
        self.truncation_length = truncation_length
        self.id_name = id_name
        self.xml = xml
        self.filter_field = filter_field
        self.filter_value = filter_value
        self.remove_empty_schema_columns = remove_empty_schema_columns
        self.seen_paths = set()
        self.persist = persist
        self.convert_flags = convert_flags

        if schema_parser:
            # schema parser does not make sheets that are persistent,
            # so use from_sheets which deep copies everything in it.
            self.main_sheet = PersistentSheet.from_sheet(
                schema_parser.main_sheet, self.connection
            )
            for sheet_name, sheet in list(self.sub_sheets.items()):
                self.sub_sheets[sheet_name] = PersistentSheet.from_sheet(
                    sheet, self.connection
                )

            self.sub_sheets = copy.deepcopy(schema_parser.sub_sheets)
            if remove_empty_schema_columns:
                # Don't use columns from the schema parser
                # (avoids empty columns)
                self.main_sheet.columns = []
                for sheet_name, sheet in list(self.sub_sheets.items()):
                    sheet.columns = []
            self.schema_parser = schema_parser
        else:
            self.schema_parser = None

        self.rollup = False
        if rollup:
            if schema_parser and len(schema_parser.rollup) > 0:
                # If rollUp is present in the schema this takes precedence over direct input.
                self.rollup = schema_parser.rollup
                if isinstance(rollup, (list,)) and (
                    len(rollup) > 1 or (len(rollup) == 1 and rollup[0] is not True)
                ):
                    warn(
                        _("Using rollUp values from schema, ignoring direct input."),
                        FlattenToolWarning,
                    )
            elif isinstance(rollup, (list,)):
                if len(rollup) == 1 and os.path.isfile(rollup[0]):
                    # Parse file, one json path per line.
                    rollup_from_file = set()
                    with open(rollup[0]) as rollup_file:
                        for line in rollup_file:
                            line = line.strip()
                            rollup_from_file.add(line)
                    self.rollup = rollup_from_file
                    # Rollup args passed directly at the commandline
                elif len(rollup) == 1 and rollup[0] is True:
                    warn(
                        _(
                            "No fields to rollup found (pass json path directly, as a list in a file, or via a schema)",
                            FlattenToolWarning,
                        )
                    )
                else:
                    self.rollup = set(rollup)
            else:
                warn(
                    _(
                        "Invalid value passed for rollup (pass json path directly, as a list in a file, or via a schema)",
                        FlattenToolWarning,
                    )
                )

        if self.xml:
            with codecs.open(json_filename, "rb") as xml_file:
                top_dict = xmltodict.parse(
                    xml_file,
                    force_list=(root_list_path,),
                    force_cdata=True,
                )
                # AFAICT, this should be true for *all* XML files
                assert len(top_dict) == 1
                root_json_dict = list(top_dict.values())[0]
                list_dict_consistency(root_json_dict)
            json_filename = None

        if json_filename is None and root_json_dict is None:
            raise FlattenToolValueError(
                _("Either json_filename or root_json_dict must be supplied")
            )

        if json_filename is not None and root_json_dict is not None:
            raise FlattenToolValueError(
                _("Only one of json_file or root_json_dict should be supplied")
            )

        if not json_filename:
            if self.root_list_path is None:
                self.root_json_list = root_json_dict
            else:
                self.root_json_list = path_search(
                    root_json_dict, self.root_list_path.split("/")
                )

        if preserve_fields:
            # Extract fields to be preserved from input file (one path per line)
            preserve_fields_all = []
            preserve_fields_input = []
            with open(preserve_fields) as preserve_fields_file:
                for line in preserve_fields_file:
                    line = line.strip()
                    path_fields = line.rsplit("/", 1)
                    preserve_fields_all = (
                        preserve_fields_all + path_fields + [line.rstrip("/")]
                    )
                    preserve_fields_input = preserve_fields_input + [line.rstrip("/")]

            self.preserve_fields = set(preserve_fields_all)
            self.preserve_fields_input = set(preserve_fields_input)

            try:
                input_not_in_schema = set()
                for field in self.preserve_fields_input:
                    if field not in self.schema_parser.flattened.keys():
                        input_not_in_schema.add(field)
                warn(
                    _(
                        "You wanted to preserve the following fields which are not present in the supplied schema: {}"
                    ).format(list(input_not_in_schema)),
                    FlattenToolWarning,
                )
            except AttributeError:
                # no schema
                pass
        else:
            self.preserve_fields = None
            self.preserve_fields_input = None

        if json_filename:
            if self.root_list_path is None:
                path = "item"
            else:
                path = root_list_path.replace("/", ".") + ".item"

            json_file = codecs.open(json_filename, encoding="utf-8")

            self.root_json_list = ijson.items(json_file, path, map_type=OrderedDict)

        try:
            self.parse()
        except ijson.common.IncompleteJSONError as err:
            raise BadlyFormedJSONError(*err.args)
        except UnicodeDecodeError as err:
            raise BadlyFormedJSONErrorUTF8(*err.args)
        finally:
            if json_filename:
                json_file.close()

    def parse(self):
        for num, json_dict in enumerate(self.root_json_list):
            if json_dict is None:
                # This is particularly useful for IATI XML, in order to not
                # fall over on empty activity, e.g. <iati-activity/>
                continue

            if not isinstance(json_dict, dict):
                warn(
                    _(f"The value at index {num} is not a JSON object"),
                    DataErrorWarning,
                )
                continue

            self.parse_json_dict(json_dict, sheet=self.main_sheet)
            # only persist every 2000 objects. peristing more often slows down storing.
            # 2000 top level objects normally not too much to store in memory.
            if num % 2000 == 0 and num != 0:
                transaction.commit()

        # This commit could be removed which would mean that upto 2000 objects
        # could be stored in memory without anything being persisted.
        transaction.commit()

        if self.remove_empty_schema_columns:
            # Remove sheets with no lines of data
            for sheet_name, sheet in list(self.sub_sheets.items()):
                if not sheet.lines:
                    del self.sub_sheets[sheet_name]

        if self.preserve_fields_input:
            nonexistent_input_paths = []
            for field in self.preserve_fields_input:
                if field not in self.seen_paths:
                    nonexistent_input_paths.append(field)
            if len(nonexistent_input_paths) > 0:
                warn(
                    _(
                        "You wanted to preserve the following fields which are not present in the input data: {}"
                    ).format(nonexistent_input_paths),
                    FlattenToolWarning,
                )

    def parse_json_dict(
        self,
        json_dict,
        sheet,
        json_key=None,
        parent_name="",
        flattened_dict=None,
        parent_id_fields=None,
        top_level_of_sub_sheet=False,
    ):
        """
        Parse a json dictionary.

        json_dict - the json dictionary
        sheet - a sheet.Sheet object representing the resulting spreadsheet
        json_key - the key that maps to this JSON dict, either directly to the dict, or to a dict that this list contains.  Is None if this dict is contained in root_json_list directly.
        """
        # Possibly main_sheet should be main_sheet_columns, but this is
        # currently named for consistency with schema.py

        if self.use_titles:
            sheet_key = sheet_key_title
        else:
            sheet_key = sheet_key_field

        skip_type_and_coordinates = False
        if (
            self.convert_flags.get("wkt")
            and "type" in json_dict
            and "coordinates" in json_dict
        ):
            if SHAPELY_LIBRARY_AVAILABLE:
                _sheet_key = sheet_key(sheet, parent_name.strip("/"))
                try:
                    geom = shapely.geometry.shape(json_dict)
                except (shapely.errors.GeometryTypeError, TypeError, ValueError) as e:
                    warn(
                        _("Invalid GeoJSON: {parser_msg}").format(parser_msg=repr(e)),
                        DataErrorWarning,
                    )
                    return
                flattened_dict[_sheet_key] = geom.wkt
                skip_type_and_coordinates = True
            else:
                warn(
                    "Install flattentool's optional geo dependencies to use geo features.",
                    FlattenToolWarning,
                )

        parent_id_fields = copy.copy(parent_id_fields) or OrderedDict()
        if flattened_dict is None:
            flattened_dict = {}
            top = True
        else:
            top = False

        if parent_name == "" and self.filter_field and self.filter_value:
            if self.filter_field not in json_dict:
                return
            if json_dict[self.filter_field] != self.filter_value:
                return

        if top_level_of_sub_sheet:
            # Add the IDs for the top level of object in an array
            for k, v in parent_id_fields.items():
                if self.xml:
                    flattened_dict[sheet_key(sheet, k)] = v["#text"]
                else:
                    flattened_dict[sheet_key(sheet, k)] = v

        if self.root_id and self.root_id in json_dict:
            parent_id_fields[sheet_key(sheet, self.root_id)] = json_dict[self.root_id]

        if self.id_name in json_dict:
            parent_id_fields[sheet_key(sheet, parent_name + self.id_name)] = json_dict[
                self.id_name
            ]

        for key, value in json_dict.items():

            if skip_type_and_coordinates and key in ["type", "coordinates"]:
                continue

            # Keep a unique list of all the JSON paths in the data that have been seen.
            parent_path = parent_name.replace("/0", "")
            full_path = parent_path + key
            self.seen_paths.add(full_path)

            if self.preserve_fields:

                siblings = False
                for field in self.preserve_fields:
                    if parent_path in field:
                        siblings = True
                if siblings and full_path not in self.preserve_fields:
                    continue

            if type(value) in BASIC_TYPES:
                if self.xml and key == "#text":
                    # Handle the text output from xmltodict
                    key = ""
                    parent_name = parent_name.strip("/")
                flattened_dict[sheet_key(sheet, parent_name + key)] = value
            elif hasattr(value, "items"):
                self.parse_json_dict(
                    value,
                    sheet=sheet,
                    json_key=key,
                    parent_name=parent_name + key + "/",
                    flattened_dict=flattened_dict,
                    parent_id_fields=parent_id_fields,
                )
            elif hasattr(value, "__iter__"):
                if all(type(x) in BASIC_TYPES for x in value):
                    # Check for an array of BASIC types
                    # TODO Make this check the schema
                    # TODO Error if the any of the values contain the separator
                    flattened_dict[sheet_key(sheet, parent_name + key)] = ";".join(
                        map(str, value)
                    )
                # Arrays of arrays
                elif all(
                    l not in BASIC_TYPES
                    and not hasattr(l, "items")
                    and hasattr(l, "__iter__")
                    and all(type(x) in BASIC_TYPES for x in l)
                    for l in value
                ):
                    flattened_dict[sheet_key(sheet, parent_name + key)] = ";".join(
                        map(lambda l: ",".join(map(str, l)), value)
                    )
                else:
                    if (
                        self.rollup and parent_name == ""
                    ):  # Rollup only currently possible to main sheet

                        if self.use_titles and not self.schema_parser:
                            warn(
                                _(
                                    "Warning: No schema was provided so column headings are JSON keys, not titles.",
                                    FlattenToolWarning,
                                )
                            )

                        if len(value) == 1:
                            for k, v in value[0].items():

                                if (
                                    self.preserve_fields
                                    and parent_name + key + "/" + k
                                    not in self.preserve_fields
                                ):
                                    continue

                                if type(v) not in BASIC_TYPES:
                                    raise FlattenToolValueError(
                                        _("Rolled up values must be basic types")
                                    )
                                else:
                                    if self.schema_parser:
                                        # We want titles and there's a schema and rollUp is in it
                                        if (
                                            self.use_titles
                                            and parent_name + key + "/0/" + k
                                            in self.schema_parser.main_sheet.titles
                                        ):
                                            flattened_dict[
                                                sheet_key_title(
                                                    sheet, parent_name + key + "/0/" + k
                                                )
                                            ] = v

                                        # We want titles and there's a schema but rollUp isn't in it
                                        # so the titles for rollup properties aren't in the main sheet
                                        # so we need to try to get the titles from a subsheet
                                        elif (
                                            self.use_titles
                                            and parent_name + key in self.rollup
                                            and self.schema_parser.sub_sheet_titles.get(
                                                (
                                                    parent_name,
                                                    key,
                                                )
                                            )
                                            in self.schema_parser.sub_sheets
                                        ):
                                            relevant_subsheet = self.schema_parser.sub_sheets.get(
                                                self.schema_parser.sub_sheet_titles.get(
                                                    (
                                                        parent_name,
                                                        key,
                                                    )
                                                )
                                            )
                                            if relevant_subsheet is not None:
                                                rollup_field_title = sheet_key_title(
                                                    relevant_subsheet,
                                                    parent_name + key + "/0/" + k,
                                                )
                                                flattened_dict[
                                                    sheet_key(sheet, rollup_field_title)
                                                ] = v

                                        # We don't want titles even though there's a schema
                                        elif not self.use_titles and (
                                            parent_name + key + "/0/" + k
                                            in self.schema_parser.main_sheet
                                            or parent_name + key in self.rollup
                                        ):
                                            flattened_dict[
                                                sheet_key(
                                                    sheet, parent_name + key + "/0/" + k
                                                )
                                            ] = v

                                    # No schema, so no titles
                                    elif parent_name + key in self.rollup:
                                        flattened_dict[
                                            sheet_key(
                                                sheet, parent_name + key + "/0/" + k
                                            )
                                        ] = v

                        elif len(value) > 1:
                            for k in set(sum((list(x.keys()) for x in value), [])):

                                if (
                                    self.preserve_fields
                                    and parent_name + key + "/" + k
                                    not in self.preserve_fields
                                ):
                                    continue

                                if (
                                    self.schema_parser
                                    and parent_name + key + "/0/" + k
                                    in self.schema_parser.main_sheet
                                ):
                                    warn(
                                        _(
                                            'More than one value supplied for "{}". Could not provide rollup, so adding a warning to the relevant cell(s) in the spreadsheet.'
                                        ).format(parent_name + key),
                                        FlattenToolWarning,
                                    )
                                    flattened_dict[
                                        sheet_key(sheet, parent_name + key + "/0/" + k)
                                    ] = _(
                                        "WARNING: More than one value supplied, consult the relevant sub-sheet for the data."
                                    )
                                elif parent_name + key in self.rollup:
                                    warn(
                                        _(
                                            'More than one value supplied for "{}". Could not provide rollup, so adding a warning to the relevant cell(s) in the spreadsheet.'
                                        ).format(parent_name + key),
                                        FlattenToolWarning,
                                    )
                                    flattened_dict[
                                        sheet_key(sheet, parent_name + key + "/0/" + k)
                                    ] = _(
                                        "WARNING: More than one value supplied, consult the relevant sub-sheet for the data."
                                    )

                    if (
                        self.use_titles
                        and self.schema_parser
                        and (
                            parent_name,
                            key,
                        )
                        in self.schema_parser.sub_sheet_titles
                    ):
                        sub_sheet_name = self.schema_parser.sub_sheet_titles[
                            (
                                parent_name,
                                key,
                            )
                        ]
                    else:
                        sub_sheet_name = make_sub_sheet_name(
                            parent_name, key, truncation_length=self.truncation_length
                        )
                    if sub_sheet_name not in self.sub_sheets:
                        self.sub_sheets[sub_sheet_name] = PersistentSheet(
                            name=sub_sheet_name, connection=self.connection
                        )

                    for json_dict in value:
                        if json_dict is None:
                            continue
                        self.parse_json_dict(
                            json_dict,
                            sheet=self.sub_sheets[sub_sheet_name],
                            json_key=key,
                            parent_id_fields=parent_id_fields,
                            parent_name=parent_name + key + "/0/",
                            top_level_of_sub_sheet=True,
                        )
            else:
                raise FlattenToolValueError(
                    _("Unsupported type {}").format(type(value))
                )

        if top:
            sheet.append_line(flattened_dict)

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        if self.persist:
            self.connection.close()
            self.db.close()
            os.remove(self.zodb_db_location)
            os.remove(self.zodb_db_location + ".lock")
            os.remove(self.zodb_db_location + ".index")
            os.remove(self.zodb_db_location + ".tmp")

1	"""
2
3	This file contains code that takes an instance of a JSON file as input (not a
4	JSON schema, for that see schema.py).
5
6	"""
7
8	import codecs	12✔
9	import copy	12✔
10	import os	12✔
11	import tempfile	12✔
12	import uuid	12✔
13	from collections import OrderedDict	12✔
14	from decimal import Decimal	12✔
15	from warnings import warn	12✔
16
17	import BTrees.OOBTree	12✔
18	import ijson	12✔
19
20	try:	12✔
21	import shapely.wkt	12✔
22
23	SHAPELY_LIBRARY_AVAILABLE = True	12✔
24	except ImportError:	12✔
25	SHAPELY_LIBRARY_AVAILABLE = False	12✔
26	import transaction	12✔
27	import xmltodict	12✔
28	import zc.zlibstorage	12✔
29	import ZODB.FileStorage	12✔
30
31	from flattentool.exceptions import (	12✔
32	DataErrorWarning,
33	FlattenToolError,
34	FlattenToolValueError,
35	FlattenToolWarning,
36	)
37	from flattentool.i18n import _	12✔
38	from flattentool.input import path_search	12✔
39	from flattentool.schema import make_sub_sheet_name	12✔
40	from flattentool.sheet import PersistentSheet	12✔
41
42	BASIC_TYPES = [str, bool, int, Decimal, type(None)]	12✔
43
44
45	class BadlyFormedJSONError(FlattenToolError, ValueError):	12✔
46	pass	12✔
47
48
49	class BadlyFormedJSONErrorUTF8(BadlyFormedJSONError):	12✔
50	pass	12✔
51
52
53	def sheet_key_field(sheet, key):	12✔
54	if key not in sheet:	12✔
55	sheet.append(key)	12✔
56	return key	12✔
57
58
59	def sheet_key_title(sheet, key):	12✔
60	"""
61	If the key has a corresponding title, return that. If doesn't, create it in the sheet and return it.
62
63	"""
64	if key in sheet.titles:	12✔
65	title = sheet.titles[key]	12✔
66	if title not in sheet:	12✔
UNCOV 67	sheet.append(title)	×
68	return title	12✔
69	else:
70	if key not in sheet:	12✔
71	sheet.append(key)	12✔
72	return key	12✔
73
74
75	def lists_of_dicts_paths(xml_dict):	12✔
76	for key, value in xml_dict.items():	12✔
77	if isinstance(value, list) and value and isinstance(value[0], dict):	12✔
78	yield (key,)	12✔
79	for x in value:	12✔
80	if isinstance(x, dict):	12✔
81	for path in lists_of_dicts_paths(x):	12✔
82	yield (key,) + path	12✔
83	elif isinstance(value, dict):	12✔
84	for path in lists_of_dicts_paths(value):	12✔
85	yield (key,) + path	12✔
86
87
88	def dicts_to_list_of_dicts(lists_of_dicts_paths_set, xml_dict, path=()):	12✔
89	for key, value in xml_dict.items():	12✔
90	if isinstance(value, list):	12✔
91	for x in value:	12✔
92	if isinstance(x, dict):	12✔
93	dicts_to_list_of_dicts(lists_of_dicts_paths_set, x, path + (key,))	12✔
94	elif isinstance(value, dict):	12✔
95	child_path = path + (key,)	12✔
96	dicts_to_list_of_dicts(lists_of_dicts_paths_set, value, child_path)	12✔
97	if child_path in lists_of_dicts_paths_set:	12✔
98	xml_dict[key] = [value]	12✔
99
100
101	def list_dict_consistency(xml_dict):	12✔
102	"""
103	For use with XML files opened with xmltodict.
104
105	If there is only one tag, xmltodict produces a dict. If there are
106	multiple, xmltodict produces a list of dicts. This functions replaces
107	dicts with lists of dicts, if there exists a list of dicts for the same
108	path elsewhere in the file.
109	"""
110	lists_of_dicts_paths_set = set(lists_of_dicts_paths(xml_dict))	12✔
111	dicts_to_list_of_dicts(lists_of_dicts_paths_set, xml_dict)	12✔
112
113
114	class JSONParser(object):	12✔
115	# Named for consistency with schema.SchemaParser, but not sure it's the most appropriate name.
116	# Similarly with methods like parse_json_dict
117
118	def __init__(	12✔
119	self,
120	json_filename=None,
121	root_json_dict=None,
122	schema_parser=None,
123	root_list_path=None,
124	root_id="ocid",
125	use_titles=False,
126	xml=False,
127	id_name="id",
128	filter_field=None,
129	filter_value=None,
130	preserve_fields=None,
131	remove_empty_schema_columns=False,
132	rollup=False,
133	truncation_length=3,
134	persist=False,
135	convert_flags={},
136	):
137	if persist:	12✔
138	# Use temp directories in OS agnostic way
139	self.zodb_db_location = (	12✔
140	tempfile.gettempdir() + "/flattentool-" + str(uuid.uuid4())
141	)
142	# zlibstorage lowers disk usage by a lot at very small performance cost
143	zodb_storage = zc.zlibstorage.ZlibStorage(	12✔
144	ZODB.FileStorage.FileStorage(self.zodb_db_location)
145	)
146	self.db = ZODB.DB(zodb_storage)	12✔
147	else:
148	# If None, in memory storage is used.
149	self.db = ZODB.DB(None)	12✔
150
151	self.connection = self.db.open()	12✔
152
153	# ZODB root, only objects attached here will be persisted
154	root = self.connection.root	12✔
155	# OOBTree means a btree with keys and values are objects (including strings)
156	root.sheet_store = BTrees.OOBTree.BTree()	12✔
157
158	self.sub_sheets = {}	12✔
159	self.main_sheet = PersistentSheet(connection=self.connection, name="")	12✔
160	self.root_list_path = root_list_path	12✔
161	self.root_id = root_id	12✔
162	self.use_titles = use_titles	12✔
163	self.truncation_length = truncation_length	12✔
164	self.id_name = id_name	12✔
165	self.xml = xml	12✔
166	self.filter_field = filter_field	12✔
167	self.filter_value = filter_value	12✔
168	self.remove_empty_schema_columns = remove_empty_schema_columns	12✔
169	self.seen_paths = set()	12✔
170	self.persist = persist	12✔
171	self.convert_flags = convert_flags	12✔
172
173	if schema_parser:	12✔
174	# schema parser does not make sheets that are persistent,
175	# so use from_sheets which deep copies everything in it.
176	self.main_sheet = PersistentSheet.from_sheet(	12✔
177	schema_parser.main_sheet, self.connection
178	)
179	for sheet_name, sheet in list(self.sub_sheets.items()):	12✔
UNCOV 180	self.sub_sheets[sheet_name] = PersistentSheet.from_sheet(	×
181	sheet, self.connection
182	)
183
184	self.sub_sheets = copy.deepcopy(schema_parser.sub_sheets)	12✔
185	if remove_empty_schema_columns:	12✔
186	# Don't use columns from the schema parser
187	# (avoids empty columns)
188	self.main_sheet.columns = []	12✔
189	for sheet_name, sheet in list(self.sub_sheets.items()):	12✔
190	sheet.columns = []	12✔
191	self.schema_parser = schema_parser	12✔
192	else:
193	self.schema_parser = None	12✔
194
195	self.rollup = False	12✔
196	if rollup:	12✔
197	if schema_parser and len(schema_parser.rollup) > 0:	12✔
198	# If rollUp is present in the schema this takes precedence over direct input.
199	self.rollup = schema_parser.rollup	12✔
200	if isinstance(rollup, (list,)) and (	12✔
201	len(rollup) > 1 or (len(rollup) == 1 and rollup[0] is not True)
202	):
UNCOV 203	warn(	×
204	_("Using rollUp values from schema, ignoring direct input."),
205	FlattenToolWarning,
206	)
207	elif isinstance(rollup, (list,)):	12✔
208	if len(rollup) == 1 and os.path.isfile(rollup[0]):	12✔
209	# Parse file, one json path per line.
210	rollup_from_file = set()	12✔
211	with open(rollup[0]) as rollup_file:	12✔
212	for line in rollup_file:	12✔
213	line = line.strip()	12✔
214	rollup_from_file.add(line)	12✔
215	self.rollup = rollup_from_file	12✔
216	# Rollup args passed directly at the commandline
217	elif len(rollup) == 1 and rollup[0] is True:	12✔
UNCOV 218	warn(	×
219	_(
220	"No fields to rollup found (pass json path directly, as a list in a file, or via a schema)",
221	FlattenToolWarning,
222	)
223	)
224	else:
225	self.rollup = set(rollup)	12✔
226	else:
UNCOV 227	warn(	×
228	_(
229	"Invalid value passed for rollup (pass json path directly, as a list in a file, or via a schema)",
230	FlattenToolWarning,
231	)
232	)
233
234	if self.xml:	12✔
235	with codecs.open(json_filename, "rb") as xml_file:	12✔
236	top_dict = xmltodict.parse(	12✔
237	xml_file,
238	force_list=(root_list_path,),
239	force_cdata=True,
240	)
241	# AFAICT, this should be true for all XML files
242	assert len(top_dict) == 1	12✔
243	root_json_dict = list(top_dict.values())[0]	12✔
244	list_dict_consistency(root_json_dict)	12✔
245	json_filename = None	12✔
246
247	if json_filename is None and root_json_dict is None:	12✔
248	raise FlattenToolValueError(	12✔
249	_("Either json_filename or root_json_dict must be supplied")
250	)
251
252	if json_filename is not None and root_json_dict is not None:	12✔
253	raise FlattenToolValueError(	12✔
254	_("Only one of json_file or root_json_dict should be supplied")
255	)
256
257	if not json_filename:	12✔
258	if self.root_list_path is None:	12✔
259	self.root_json_list = root_json_dict	12✔
260	else:
261	self.root_json_list = path_search(	12✔
262	root_json_dict, self.root_list_path.split("/")
263	)
264
265	if preserve_fields:	12✔
266	# Extract fields to be preserved from input file (one path per line)
267	preserve_fields_all = []	12✔
268	preserve_fields_input = []	12✔
269	with open(preserve_fields) as preserve_fields_file:	12✔
270	for line in preserve_fields_file:	12✔
271	line = line.strip()	12✔
272	path_fields = line.rsplit("/", 1)	12✔
273	preserve_fields_all = (	12✔
274	preserve_fields_all + path_fields + [line.rstrip("/")]
275	)
276	preserve_fields_input = preserve_fields_input + [line.rstrip("/")]	12✔
277
278	self.preserve_fields = set(preserve_fields_all)	12✔
279	self.preserve_fields_input = set(preserve_fields_input)	12✔
280
281	try:	12✔
282	input_not_in_schema = set()	12✔
283	for field in self.preserve_fields_input:	12✔
284	if field not in self.schema_parser.flattened.keys():	12✔
UNCOV 285	input_not_in_schema.add(field)	×
UNCOV 286	warn(	×
287	_(
288	"You wanted to preserve the following fields which are not present in the supplied schema: {}"
289	).format(list(input_not_in_schema)),
290	FlattenToolWarning,
291	)
292	except AttributeError:	12✔
293	# no schema
294	pass	12✔
295	else:
296	self.preserve_fields = None	12✔
297	self.preserve_fields_input = None	12✔
298
299	if json_filename:	12✔
300	if self.root_list_path is None:	12✔
301	path = "item"	12✔
302	else:
303	path = root_list_path.replace("/", ".") + ".item"	12✔
304
305	json_file = codecs.open(json_filename, encoding="utf-8")	12✔
306
307	self.root_json_list = ijson.items(json_file, path, map_type=OrderedDict)	12✔
308
309	try:	12✔
310	self.parse()	12✔
311	except ijson.common.IncompleteJSONError as err:	12✔
312	raise BadlyFormedJSONError(*err.args)	12✔
313	except UnicodeDecodeError as err:	12✔
314	raise BadlyFormedJSONErrorUTF8(*err.args)	12✔
315	finally:
316	if json_filename:	12✔
317	json_file.close()	12✔
318
319	def parse(self):	12✔
320	for num, json_dict in enumerate(self.root_json_list):	12✔
321	if json_dict is None:	12✔
322	# This is particularly useful for IATI XML, in order to not
323	# fall over on empty activity, e.g. <iati-activity/>
324	continue	12✔
325
326	if not isinstance(json_dict, dict):	12✔
327	warn(	12✔
328	_(f"The value at index {num} is not a JSON object"),
329	DataErrorWarning,
330	)
331	continue	12✔
332
333	self.parse_json_dict(json_dict, sheet=self.main_sheet)	12✔
334	# only persist every 2000 objects. peristing more often slows down storing.
335	# 2000 top level objects normally not too much to store in memory.
336	if num % 2000 == 0 and num != 0:	12✔
UNCOV 337	transaction.commit()	×
338
339	# This commit could be removed which would mean that upto 2000 objects
340	# could be stored in memory without anything being persisted.
341	transaction.commit()	12✔
342
343	if self.remove_empty_schema_columns:	12✔
344	# Remove sheets with no lines of data
345	for sheet_name, sheet in list(self.sub_sheets.items()):	12✔
346	if not sheet.lines:	12✔
347	del self.sub_sheets[sheet_name]	12✔
348
349	if self.preserve_fields_input:	12✔
350	nonexistent_input_paths = []	12✔
351	for field in self.preserve_fields_input:	12✔
352	if field not in self.seen_paths:	12✔
353	nonexistent_input_paths.append(field)	12✔
354	if len(nonexistent_input_paths) > 0:	12✔
355	warn(	12✔
356	_(
357	"You wanted to preserve the following fields which are not present in the input data: {}"
358	).format(nonexistent_input_paths),
359	FlattenToolWarning,
360	)
361
362	def parse_json_dict(	12✔
363	self,
364	json_dict,
365	sheet,
366	json_key=None,
367	parent_name="",
368	flattened_dict=None,
369	parent_id_fields=None,
370	top_level_of_sub_sheet=False,
371	):
372	"""
373	Parse a json dictionary.
374
375	json_dict - the json dictionary
376	sheet - a sheet.Sheet object representing the resulting spreadsheet
377	json_key - the key that maps to this JSON dict, either directly to the dict, or to a dict that this list contains. Is None if this dict is contained in root_json_list directly.
378	"""
379	# Possibly main_sheet should be main_sheet_columns, but this is
380	# currently named for consistency with schema.py
381
382	if self.use_titles:	12✔
383	sheet_key = sheet_key_title	12✔
384	else:
385	sheet_key = sheet_key_field	12✔
386
387	skip_type_and_coordinates = False	12✔
388	if (	12✔
389	self.convert_flags.get("wkt")
390	and "type" in json_dict
391	and "coordinates" in json_dict
392	):
393	if SHAPELY_LIBRARY_AVAILABLE:	12✔
394	_sheet_key = sheet_key(sheet, parent_name.strip("/"))	12✔
395	try:	12✔
396	geom = shapely.geometry.shape(json_dict)	12✔
397	except (shapely.errors.GeometryTypeError, TypeError, ValueError) as e:	12✔
398	warn(	12✔
399	_("Invalid GeoJSON: {parser_msg}").format(parser_msg=repr(e)),
400	DataErrorWarning,
401	)
402	return	12✔
403	flattened_dict[_sheet_key] = geom.wkt	12✔
404	skip_type_and_coordinates = True	12✔
405	else:
UNCOV 406	warn(	×
407	"Install flattentool's optional geo dependencies to use geo features.",
408	FlattenToolWarning,
409	)
410
411	parent_id_fields = copy.copy(parent_id_fields) or OrderedDict()	12✔
412	if flattened_dict is None:	12✔
413	flattened_dict = {}	12✔
414	top = True	12✔
415	else:
416	top = False	12✔
417
418	if parent_name == "" and self.filter_field and self.filter_value:	12✔
419	if self.filter_field not in json_dict:	12✔
UNCOV 420	return	×
421	if json_dict[self.filter_field] != self.filter_value:	12✔
422	return	12✔
423
424	if top_level_of_sub_sheet:	12✔
425	# Add the IDs for the top level of object in an array
426	for k, v in parent_id_fields.items():	12✔
427	if self.xml:	12✔
428	flattened_dict[sheet_key(sheet, k)] = v["#text"]	12✔
429	else:
430	flattened_dict[sheet_key(sheet, k)] = v	12✔
431
432	if self.root_id and self.root_id in json_dict:	12✔
433	parent_id_fields[sheet_key(sheet, self.root_id)] = json_dict[self.root_id]	12✔
434
435	if self.id_name in json_dict:	12✔
436	parent_id_fields[sheet_key(sheet, parent_name + self.id_name)] = json_dict[	12✔
437	self.id_name
438	]
439
440	for key, value in json_dict.items():	12✔
441
442	if skip_type_and_coordinates and key in ["type", "coordinates"]:	12✔
443	continue	12✔
444
445	# Keep a unique list of all the JSON paths in the data that have been seen.
446	parent_path = parent_name.replace("/0", "")	12✔
447	full_path = parent_path + key	12✔
448	self.seen_paths.add(full_path)	12✔
449
450	if self.preserve_fields:	12✔
451
452	siblings = False	12✔
453	for field in self.preserve_fields:	12✔
454	if parent_path in field:	12✔
455	siblings = True	12✔
456	if siblings and full_path not in self.preserve_fields:	12✔
457	continue	12✔
458
459	if type(value) in BASIC_TYPES:	12✔
460	if self.xml and key == "#text":	12✔
461	# Handle the text output from xmltodict
462	key = ""	12✔
463	parent_name = parent_name.strip("/")	12✔
464	flattened_dict[sheet_key(sheet, parent_name + key)] = value	12✔
465	elif hasattr(value, "items"):	12✔
466	self.parse_json_dict(	12✔
467	value,
468	sheet=sheet,
469	json_key=key,
470	parent_name=parent_name + key + "/",
471	flattened_dict=flattened_dict,
472	parent_id_fields=parent_id_fields,
473	)
474	elif hasattr(value, "__iter__"):	12✔
475	if all(type(x) in BASIC_TYPES for x in value):	12✔
476	# Check for an array of BASIC types
477	# TODO Make this check the schema
478	# TODO Error if the any of the values contain the separator
479	flattened_dict[sheet_key(sheet, parent_name + key)] = ";".join(	12✔
480	map(str, value)
481	)
482	# Arrays of arrays
483	elif all(	12✔
484	l not in BASIC_TYPES
485	and not hasattr(l, "items")
486	and hasattr(l, "__iter__")
487	and all(type(x) in BASIC_TYPES for x in l)
488	for l in value
489	):
490	flattened_dict[sheet_key(sheet, parent_name + key)] = ";".join(	12✔
491	map(lambda l: ",".join(map(str, l)), value)
492	)
493	else:
494	if (	12✔
495	self.rollup and parent_name == ""
496	): # Rollup only currently possible to main sheet
497
498	if self.use_titles and not self.schema_parser:	12✔
UNCOV 499	warn(	×
500	_(
501	"Warning: No schema was provided so column headings are JSON keys, not titles.",
502	FlattenToolWarning,
503	)
504	)
505
506	if len(value) == 1:	12✔
507	for k, v in value[0].items():	12✔
508
509	if (	12✔
510	self.preserve_fields
511	and parent_name + key + "/" + k
512	not in self.preserve_fields
513	):
UNCOV 514	continue	×
515
516	if type(v) not in BASIC_TYPES:	12✔
NEW 517	raise FlattenToolValueError(	×
518	_("Rolled up values must be basic types")
519	)
520	else:
521	if self.schema_parser:	12✔
522	# We want titles and there's a schema and rollUp is in it
523	if (	12✔
524	self.use_titles
525	and parent_name + key + "/0/" + k
526	in self.schema_parser.main_sheet.titles
527	):
528	flattened_dict[	12✔
529	sheet_key_title(
530	sheet, parent_name + key + "/0/" + k
531	)
532	] = v
533
534	# We want titles and there's a schema but rollUp isn't in it
535	# so the titles for rollup properties aren't in the main sheet
536	# so we need to try to get the titles from a subsheet
537	elif (	12✔
538	self.use_titles
539	and parent_name + key in self.rollup
540	and self.schema_parser.sub_sheet_titles.get(
541	(
542	parent_name,
543	key,
544	)
545	)
546	in self.schema_parser.sub_sheets
547	):
548	relevant_subsheet = self.schema_parser.sub_sheets.get(	12✔
549	self.schema_parser.sub_sheet_titles.get(
550	(
551	parent_name,
552	key,
553	)
554	)
555	)
556	if relevant_subsheet is not None:	12✔
557	rollup_field_title = sheet_key_title(	12✔
558	relevant_subsheet,
559	parent_name + key + "/0/" + k,
560	)
561	flattened_dict[	12✔
562	sheet_key(sheet, rollup_field_title)
563	] = v
564
565	# We don't want titles even though there's a schema
566	elif not self.use_titles and (	12✔
567	parent_name + key + "/0/" + k
568	in self.schema_parser.main_sheet
569	or parent_name + key in self.rollup
570	):
571	flattened_dict[	12✔
572	sheet_key(
573	sheet, parent_name + key + "/0/" + k
574	)
575	] = v
576
577	# No schema, so no titles
578	elif parent_name + key in self.rollup:	12✔
579	flattened_dict[	12✔
580	sheet_key(
581	sheet, parent_name + key + "/0/" + k
582	)
583	] = v
584
585	elif len(value) > 1:	12✔
586	for k in set(sum((list(x.keys()) for x in value), [])):	12✔
587
588	if (	12✔
589	self.preserve_fields
590	and parent_name + key + "/" + k
591	not in self.preserve_fields
592	):
UNCOV 593	continue	×
594
595	if (	12✔
596	self.schema_parser
597	and parent_name + key + "/0/" + k
598	in self.schema_parser.main_sheet
599	):
600	warn(	12✔
601	_(
602	'More than one value supplied for "{}". Could not provide rollup, so adding a warning to the relevant cell(s) in the spreadsheet.'
603	).format(parent_name + key),
604	FlattenToolWarning,
605	)
606	flattened_dict[	12✔
607	sheet_key(sheet, parent_name + key + "/0/" + k)
608	] = _(
609	"WARNING: More than one value supplied, consult the relevant sub-sheet for the data."
610	)
611	elif parent_name + key in self.rollup:	12✔
UNCOV 612	warn(	×
613	_(
614	'More than one value supplied for "{}". Could not provide rollup, so adding a warning to the relevant cell(s) in the spreadsheet.'
615	).format(parent_name + key),
616	FlattenToolWarning,
617	)
UNCOV 618	flattened_dict[	×
619	sheet_key(sheet, parent_name + key + "/0/" + k)
620	] = _(
621	"WARNING: More than one value supplied, consult the relevant sub-sheet for the data."
622	)
623
624	if (	12✔
625	self.use_titles
626	and self.schema_parser
627	and (
628	parent_name,
629	key,
630	)
631	in self.schema_parser.sub_sheet_titles
632	):
633	sub_sheet_name = self.schema_parser.sub_sheet_titles[	12✔
634	(
635	parent_name,
636	key,
637	)
638	]
639	else:
640	sub_sheet_name = make_sub_sheet_name(	12✔
641	parent_name, key, truncation_length=self.truncation_length
642	)
643	if sub_sheet_name not in self.sub_sheets:	12✔
644	self.sub_sheets[sub_sheet_name] = PersistentSheet(	12✔
645	name=sub_sheet_name, connection=self.connection
646	)
647
648	for json_dict in value:	12✔
649	if json_dict is None:	12✔
650	continue	12✔
651	self.parse_json_dict(	12✔
652	json_dict,
653	sheet=self.sub_sheets[sub_sheet_name],
654	json_key=key,
655	parent_id_fields=parent_id_fields,
656	parent_name=parent_name + key + "/0/",
657	top_level_of_sub_sheet=True,
658	)
659	else:
NEW 660	raise FlattenToolValueError(	×
661	_("Unsupported type {}").format(type(value))
662	)
663
664	if top:	12✔
665	sheet.append_line(flattened_dict)	12✔
666
667	def __enter__(self):	12✔
668	return self	12✔
669
670	def __exit__(self, type, value, traceback):	12✔
671	if self.persist:	12✔
672	self.connection.close()	12✔
673	self.db.close()	12✔
674	os.remove(self.zodb_db_location)	12✔
675	os.remove(self.zodb_db_location + ".lock")	12✔
676	os.remove(self.zodb_db_location + ".index")	12✔
677	os.remove(self.zodb_db_location + ".tmp")	12✔

OpenDataServices / flatten-tool / 10202339569

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