Загрузка ФИАС в БД

fiasmodel.py

from functools import partial
from io import TextIOWrapper
from os import pipe, fdopen
from threading import Thread
from typing import Tuple

from lxml.etree import _Element, XMLPullParser

from app import app
from modules.core.db import BaseModel, db


class FiasModel(BaseModel):
    """
    Модель, которую мы загружаем из базы ФИАС.
    """
    @classmethod
    def load_xml(cls, file: TextIOWrapper, fail_silently: bool = False):
        """
        Создаёт таблицу и загружает данные из XML-файла.
        :param file: Открытый поток XML-файла из состава ФИАС.
        :param fail_silently: Параметр, с которым будет вызвана create_table() в начале.
        """
        cls.create_table(fail_silently=fail_silently)
        r_fd, w_fd = pipe()
        
        # Делаем функцию, которая будет сидеть на конце пайпа и кормить его содержимым постгрес
        def bulk():
            with fdopen(r_fd, "r") as reader:
                tablename = cls._meta.schema + "." + cls._meta.db_table
                columns = cls._meta.sorted_field_names
                db.get_conn().cursor().copy_from(reader, tablename, columns=columns)
                db.get_conn().cursor().execute("commit")
        
        # Запускаем ту функцию в отдельном потоке
        reading_thread = Thread(target=bulk)
        reading_thread.start()
        
        # А сами тем временем открываем исходный файл и переливаем его содержимое
        # из XML в текстовый формат для постгреса, в котором и отправляем в пайп
        try:
            with fdopen(w_fd, "w") as writer:
                # Первые байты в ФИАСовских файлах какие-то странные, пропускаем их
                file.read(3)
                
                # Читаем файл по кусочкам, пока он не перестанет читаться
                xml = XMLPullParser(events=["end"])
                for chunk in iter(partial(file.read, 100*1024), ""):
                    if len(chunk) == 0:
                        break
                    
                    xml.feed(chunk)
                    
                    # Читаем всё свежевозникшие события
                    for event, elem in xml.read_events():  # type: Tuple[str,_Element]
                        try:
                            # Каждый подкласс FiasModel используется только для поиска определённого вида тэгов.
                            # Остальные тэги игнорируем.
                            if elem.tag != cls._meta.fias_tag:
                                continue
                            
                            # Собираем строку из полей в порядке их объявления в модели.
                            # Отсутствующие поля пишем как \N. Между полями ставим знаки табуляции.
                            # В конце добавляем перенос строки.
                            fields = cls._meta.sorted_field_names
                            line = "\t".join(elem.attrib.get(field.upper(), "\\N") for field in fields)
                            writer.write(line + "\n")
                        
                        finally:
                            # Обязательно удаляем ноду, чтобы не занимала память
                            # (если у ноды нет родителя, то удалить её не получится, да и не нужно)
                            try:
                                elem.getparent().remove(elem)
                            except AttributeError:
                                pass
        
        finally:
            # Ждём, пока закончит свою работу тред с постгресом
            reading_thread.join()

fiasrar.py

class FiasRar:
    """
    Обёртка над RAR-архивом от ФИАСа, позволяющая читать из него данные.
    Полагается на наличие в системе утилиты unrar, причём именно несвободной её реализации,
    так как в unrar-free отсутствует возможность получать содержимое файла пайпом.
    """
    def __init__(self, archive_name: str):
        self.archive_name = archive_name
    
    
    def list_files(self) -> Iterable[str]:
        """
        Лениво возвращает список всех файлов в архиве.
        """
        unrar = Popen([app.config.bin.unrar, "lb", self.archive_name], stdout=PIPE)
        for line in unrar.stdout.readlines():
            yield line.decode().strip()
    
    
    def get_file(self, filename: str) -> TextIOWrapper:
        """
        Возвращает файлоподобный объект по его точному имени.
        """
        unrar = Popen([app.config.bin.unrar, "p", "-inul", self.archive_name, filename], stdout=PIPE)
        return unrar.stdout
    
    
    def get_file_by_prefix(self, prefix: str) -> TextIOWrapper:
        """
        Возвращает файлоподобный объект по началу его имени,
        например, AS_HOUSE вместо AS_HOUSE_20160629_5ef74f2b-451e-4a19-aaaa-b2749816ea18.XML.
        
        Во избежание случайной путаницы между такими файлами как AR_HOUSE_* и AR_HOUSEINT_*
        к префиксу при проверке автоматически добавляется "_".
        """
        for filename in self.list_files():
            if filename.startswith(prefix+"_"):
                return self.get_file(filename)

socrbase.py

from peewee import CharField, DecimalField

from modules.geo.models.fias.fiasmodel import FiasModel


class FiasAddressObjectType(FiasModel):
    class Meta:
        schema = "fias"
        db_table = "socrbase"
        primary_key = False
        
        fias_tag = "AddressObjectType"
    
    
    kod_t_st = CharField(4)  # type: str
    """Ключевое поле"""
    
    level = DecimalField(10, 0)  # type: int
    """Уровень адресного объекта"""
    
    scname = CharField(50, null=True)  # type: str
    """Краткое наименование типа объекта"""
    
    socrname = CharField(50)  # type: str
    """Полное наименование типа объекта"""

Пример использования.py

rar = FiasRar(archive_file)
FiasAddressObjectType.load_xml(rar.get_file_by_prefix("AS_SOCRBASE"))

А не могли бы вы опубликовать BaseModel, db пожалуйста