Блог

У меня есть куча данных JSON, которые я делал в основном вручную. Несколько тысяч строк. Мне нужно преобразовать его в совершенно другой формат, используя Python.

Обзор моих «материалов»:

Столбец: основная «единица» моих данных. У каждого столбца есть атрибуты. Не беспокойтесь о значении атрибутов, но атрибуты должны быть сохранены для каждого столбца, если они существуют.

Папка: папки группируют столбцы и другие папки вместе. Папки в настоящее время не имеют атрибутов, они (в настоящее время) содержат только другие объекты папки и столбца (Объект здесь не обязательно ссылается на объекты JSON … скорее на «сущность»)

Вселенная: вселенные группируют все в большие куски, которые в большей части моего проекта не могут взаимодействовать друг с другом. Здесь это не важно, но это то, что они делают.

Некоторые ограничения:

• Столбец не может содержать другие объекты столбца, объекты папки или объекты Юниверса.
• Папки не могут содержать объекты юниверса.
• Юниверсы не могут содержать другие объекты Юниверса.

В настоящее время у меня есть столбцы в этой форме:

 "Column0Name": {
  "type": "a type",
  "dtype": "data type",
  "description": "abcdefg"
}
  

и мне нужно, чтобы он перешел к:

 {
  "name": "Column0Name",
  "type": "a type",
  "dtype": "data type",
  "description": "abcdefg"
}
  

По сути, мне нужно преобразовать элементы типа «ключ-значение столбца» в массив элементов (я новичок в JSON, не знаю терминологии). Мне также нужно, чтобы каждая папка заканчивалась двумя новыми массивами JSON (в дополнение к паре ключ-значение «имя»: «FolderName»). Для этого необходимо добавить "folders": [] и "columns": [] . Итак, у меня есть это для папок:

 "Folder0Name": {
  "Column0Name": {
    "type": "a",
    "dtype": "b",
    "description": "c"
  },
  "Column1Name": {
    "type": "d",
    "dtype": "e",
    "description": "f"
  }
}
  

и нужно перейти к этому:

 {
  "name": "Folder0Name",
  "folders": [],
  "columns": [
    {"name": "Column0Name", "type": "a", "dtype": "b", "description": "c"},
    {"name": "Column1Name", "type": "d", "dtype": "e", "description": "f"}
  ]
}
  

Папки также окажутся в массиве внутри его родительского юниверса. Аналогично, каждая вселенная будет содержать «имя», «папки» и «столбцы». Как таковой:

 {
  "name": "Universe0",
  "folders": [a bunch of folders in a JSON array],
  "columns": [occasionally some columns in a JSON array]
}
  

Bottom line:

• I’m going to guess that I need a recursive function to iterate though all the nested dictionaries after I import the JSON data with the json Python module.
• I’m thinking some sort of usage of yield might help but I’m not super familiar yet with it.
• Would it be easier to update the dict s as I go, or destroy each key-value pairs and construct an entirely new dict as I go?

Here is what I have so far. I’m stuck on getting the generator to return actual dictionaries instead of a generator object.

 import json


class AllUniverses:
    """Container to hold all the Universes found in the json file"""
    def __init__(self, filename):
        self._fn = filename
        self.data = {}
        self.read_data()

    def read_data(self):
        with open(self._fn, 'r') as fin:
            self.data = json.load(fin)
        return self

    def universe_key(self):
        """Get the next universe key from the dict of all universes

            The key will be used as the name for the universe.
        """
        yield from self.data


class Universe:
    def __init__(self, json_filename):
        self._au = AllUniverses(filename=json_filename)
        self.uni_key = self._au.universe_key()
        self._universe_data = self._au.data.copy()
        self._col_attrs = ['type', 'dtype', 'description', 'aggregation']
        self._folders_list = []
        self._columns_list = []
        self._type = "Universe"
        self._name = ""
        self.uni = dict()
        self.is_folder = False
        self.is_column = False

    def output(self):
        # TODO: Pass this to json.dump?
        # TODO: Still need to get the actual folder and column dictionaries
        #  from the generators
        out = {
            "name": self._name,
            "type": "Universe",
            "folder": [f.me for f in self._folders_list],
            "columns": [c.me for c in self._columns_list]}
        return out

    def update_universe(self):
        """Get the next universe"""
        universe_k = next(self.uni_key)
        self._name = str(universe_k)
        self.uni = self._universe_data.pop(universe_k)
        return self

    def parse_nodes(self):
        """Process all child nodes"""
        nodes = [_ for _ in self.uni.keys()]
        for k in nodes:
            v = self.uni.pop(k)
            self._is_column(val=v)
            if self.is_column:
                fc = Column(data=v, key_name=k)
                self._columns_list.append(fc)
            else:
                fc = Folder(data=v, key_name=k)
                self._folders_list.append(fc)
        return self

    def _is_column(self, val):
        """Determine if val is a Column or Folder object"""
        self.is_folder = False
        self._column = False
        if isinstance(val, dict) and not val:
            self.is_folder = True
        elif not isinstance(val, dict):
            raise TypeError('Cannot handle inputs not of type dict')
        elif any([i in val.keys() for i in self._col_attrs]):
            self._column = True
        else:
            self.is_folder = True
        return self

    def parse_children(self):
        for folder in self._folders_list:
            assert(isinstance(folder, Folder)), f'bletch idk what happened'
            folder.parse_nodes()


class Folder:
    def __init__(self, data, key_name):
        self._data = data.copy()
        self._name = str(key_name)
        self._node_keys = [_ for _ in self._data.keys()]
        self._folders = []
        self._columns = []
        self._col_attrs = ['type', 'dtype', 'description', 'aggregation']

    @property
    def me(self):
        # maybe this should force the code to parse all children of this
        # Folder? Need to convert the generator into actual dictionaries
        return {"name": self._name, "type": "Folder",
                "columns": [(c.me for c in self._columns)],
                "folders": [(f.me for f in self._folders)]}

    def parse_nodes(self):
        """Parse all the children of this Folder

            Parse through all the node names. If it is detected to be a Folder
            then create a Folder obj. from it and add to the list of Folder
            objects. Else create a Column obj. from it and append to the list
            of Column obj.

            This should be appending dictionaries
        """
        for key in self._node_keys:
            _folder = False
            _column = False
            values = self._data.copy()[key]

            if isinstance(values, dict) and not values:
                _folder = True
            elif not isinstance(values, dict):
                raise TypeError('Cannot handle inputs not of type dict')
            elif any([i in values.keys() for i in self._col_attrs]):
                _column = True
            else:
                _folder = True
            if _folder:
                f = Folder(data=values, key_name=key)
                self._folders.append(f.me)
            else:
                c = Column(data=values, key_name=key)
                self._columns.append(c.me)
        return self


class Column:
    def __init__(self, data, key_name):
        self._data = data.copy()
        self._stupid_check()
        self._me = {
            'name': str(key_name),
            'type': 'Column',
            'ctype': self._data.pop('type'),
            'dtype': self._data.pop('dtype'),
            'description': self._data.pop('description'),
            'aggregation': self._data.pop('aggregation')}

    def __str__(self):
        # TODO: pretty sure this isn't correct
        return str(self.me)

    @property
    def me(self):
        return self._me

    def to_json(self):
        # This seems to be working? I think?
        return json.dumps(self, default=lambda o: str(self.me))  # o.__dict__)

    def _stupid_check(self):
        """If the key isn't in the dictionary, add it"""
        keys = [_ for _ in self._data.keys()]
        keys_defining_a_column = ['type', 'dtype', 'description', 'aggregation']
        for json_key in keys_defining_a_column:
            if json_key not in keys:
                self._data[json_key] = ""
        return self


if __name__ == "__main__":
    file = r"dummy_json_data.json"
    u = Universe(json_filename=file)
    u.update_universe()
    u.parse_nodes()
    u.parse_children()
    print('check me')
  

И это дает мне это:

 {
    "name":"UniverseName",
    "type":"Universe",
    "folder":[
        {"name":"Folder0Name",
            "type":"Folder",
            "columns":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB0B0>],
            "folders":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB190>]
        },
        {"name":"Folder2Name",
            "type":"Folder",
            "columns":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB040>],
            "folders":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB120>]
        },
        {"name":"Folder4Name",
            "type":"Folder",
            "columns":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB270>],
            "folders":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB200>]
        },
        {"name":"Folder6Name",
            "type":"Folder",
            "columns":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB2E0>],
            "folders":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB350>]
        },
        {"name":"Folder8Name",
            "type":"Folder",
            "columns":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB3C0>],
            "folders":[<generator object Folder.me.<locals>.<genexpr> at 0x000001ACFBEDB430>]
        }
    ],
    "columns":[]
}
  

Если существует существующий инструмент для такого рода преобразования, чтобы мне не приходилось писать код на Python, это тоже было бы привлекательной альтернативой.

Давайте создадим 3 класса, необходимые для представления Column s, Folder s и Unverse s. Прежде чем приступить к некоторым темам, о которых я хочу поговорить, я даю их краткое описание здесь, если какая-либо из них для вас нова, я могу углубиться:

• Я буду использовать аннотации типов, чтобы было ясно, к какому типу относится каждая переменная.
• Я собираюсь использовать __slots__ . Сообщая Column классу, что его экземпляры будут иметь name , ctype , dtype , description и aggragation атрибуты, каждому экземпляру Column потребуется меньше места в памяти. Недостатком является то, что он не будет принимать никаких других атрибутов, не перечисленных там. Это экономит память, но теряет гибкость. Поскольку у нас будет несколько (возможно, сотни или тысячи) экземпляров, сокращение объема памяти кажется более важным, чем гибкость добавления любого атрибута.
• Каждый класс будет иметь стандартный конструктор, где каждый аргумент имеет значение по умолчанию, кроме name, которое является обязательным.
• У каждого класса будет вызываться другой конструктор from_old_syntax . Это будет метод класса, который получает строку, соответствующую имени, и dict, соответствующий данным, в качестве своих аргументов и выводит соответствующий экземпляр ( Column , Folder или Universe ).
• Universe s в основном такие же, как Folder s с разными именами (на данный момент), поэтому он в основном унаследует его ( class Universe(Folder): pass ).

 from typing import List


class Column:
    __slots__ = 'name', 'ctype', 'dtype', 'description', 'aggregation'

    def __init__(
        self,
        name: str,
        ctype: str = '',
        dtype: str = '',
        description: str = '',
        aggregation: str = '',
    ) -> None:
        self.name = name
        self.ctype = ctype
        self.dtype = dtype
        self.description = description
        self.aggregation = aggregation

    @classmethod
    def from_old_syntax(cls, name: str, data: dict) -> "Column":
        column = cls(name)
        for key, value in data.items():
            # The old syntax used type for column type but in the new syntax it
            # will have another meaning so we use ctype instead
            if key == 'type':
                key = 'ctype'
            try:
                setattr(column, key, value)
            except AttributeError as e:
                raise AttributeError(f"Unexpected key {key} for Column") from e
        return column


class Folder:
    __slots__ = 'name', 'folders', 'columns'

    def __init__(
        self,
        name: str,
        columns: List[Column] = None,
        folders: List["Folder"] = None,
    ) -> None:
        self.name = name
        if columns is None:
            self.columns = []
        else:
            self.columns = [column for column in columns]
        if folders is None:
            self.folders = []
        else:
            self.folders = [folder for folder in folders]

    @classmethod
    def from_old_syntax(cls, name: str, data: dict) -> "Folder":
        columns = []  # type: List[Column]
        folders = []  # type: List["Folder"]
        for key, value in data.items():
            # Determine if it is a Column or a Folder
            if 'type' in value and 'dtype' in value:
                columns.append(Column.from_old_syntax(key, value))
            else:
                folders.append(Folder.from_old_syntax(key, value))
        return cls(name, columns, folders)


class Universe(Folder):
    pass
  

Как вы можете видеть, конструкторы довольно тривиальны, назначьте аргументы атрибутам и готово. В случае Folder s (и, следовательно, в Universe s тоже) двумя аргументами являются списки столбцов и папок. Значение по умолчанию равно None (в данном случае мы инициализируем как пустой список), поскольку использование изменяемых переменных в качестве значений по умолчанию имеет некоторые проблемы, поэтому рекомендуется использовать None в качестве значения по умолчанию для изменяемых переменных (таких как списки).

Column from_old_syntax Метод класса создает пустой Column файл с указанным именем. После этого мы выполняем итерацию по dict данных, который также был предоставлен, и присваиваем его пару ключ-значение соответствующему атрибуту. Существует особый случай, когда ключ «type» преобразуется в «ctype», поскольку «type» будет использоваться для другой цели с новым синтаксисом. Само назначение выполняется setattr(column, key, value) . Мы включили это в try ... except ... предложение, потому что, как мы сказали выше, в качестве атрибутов могут использоваться только элементы в __slots__ , поэтому, если есть атрибут, который вы забыли, вы получите исключение с надписью «AttributeError: неожиданное имя ключа», и вам нужно будет только добавить это «ИМЯ» к __slots__ .

Folder метод класса (и, следовательно, Unverse from_old_syntax ) в Python еще проще. Создайте список столбцов и папок, выполните итерацию по данным, проверяя, является ли это папкой или столбцом, и используйте соответствующий from_old_syntax метод класса. Затем используйте эти два списка и предоставленное имя, чтобы вернуть экземпляр. Обратите внимание, что Folder.from_old_syntax обозначения используются для создания папок вместо cls.from_old_syntax потому что cls может быть Universe . Однако для создания insdance мы используем cls(...) как здесь мы хотим использовать Universe или Folder .

Теперь вы можете указать, universes = [Universe.from_old_syntax(name, data) for name, data in json.load(f).items()] где f находится файл, и вы получите все свои Universe s, Folder s и Column s в памяти. Итак, теперь нам нужно закодировать их обратно в JSON. Для этого мы собираемся расширить json.JSONEncoder , чтобы он знал, как анализировать наши классы в словари, которые он может нормально кодировать. Для этого вам просто нужно перезаписать default метод, проверить, относится ли переданный объект к нашим классам, и вернуть dict, который будет закодирован. Если это не один из наших классов, мы позволим родительскому default методу позаботиться об этом.

 import json


# JSON fields with this values will be omitted
EMPTY_VALUES = "", [], {}


class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, (Column, Folder, Universe)):
            # Make a dict with every item in their respective __slots__
            data = {
                attr: getattr(obj, attr) for attr in obj.__slots__
                if getattr(obj, attr) not in EMPTY_VALUES
            }
            # Add the type fild with the class name
            data['type'] = obj.__class__.__name__
            return data

        # Use the parent class function for any object not handled explicitly
        super().default(obj)
  

Преобразование классов в словари в основном принимает то, что находится в __slots__ , в качестве ключа и значения атрибута в качестве значения. Мы будем фильтровать те значения, которые представляют собой пустую строку, пустой список или пустой dict, поскольку нам не нужно записывать их в JSON. Наконец, мы добавляем ключ «type» в dict, читая имя класса объектов ( Column , Folder и Universe ).

Чтобы использовать его, вы должны передать CustomEncoder в качестве cls аргумента json.dump .

Таким образом, код будет выглядеть следующим образом (опуская определения классов, чтобы сделать его коротким):

 import json
from typing import List


# JSON fields with this values will be omitted
EMPTY_VALUES = "", [], {}


class Column:
    # ...


class Folder:
    # ...


class Universe(Folder):
    pass


class CustomEncoder(json.JSONEncoder):
    # ...


if __name__ == '__main__':
    with open('dummy_json_data.json', 'r') as f_in, open('output.json', 'w') as f_out:
        universes = [Universe.from_old_syntax(name, data)
                     for name, data in json.load(f_in).items()]
        json.dump(universes, f_out, cls=CustomEncoder, indent=4)