SQLAlchemy: automap_base и fork процесса

Question

Делаю сервис АПИ который работает с БД на MySQL отражая существующую схему, а также форкается в несколько процессов. Мой код по работе с БД примерно таков:

from sqlalchemy import MetaData
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm.session import Session

my_engine = create_engine_by_info(my_config)
metadata = MetaData(bind=my_engine)
Base: type = automap_base(metadata=metadata)

class User(Base):
    __tablename__ = 'auth_user'
    # Описание отношение...

# Другие классы...

Base.prepare(my_engine, reflect=True)

def find_user(field):
    with Session(my_engine) as session:
        query = session.query(User)
        query = query.filter(User.field == field)
        records = query.all()
        for u in records:
            return u
        return None

Всё работает хорошо до разделения процессов: после завершения работы ребёнка исходный процесс теряет подключение: Lost connection to MySQL server during query.

Вероятно надо иметь отдельные my_engine для каждого процесса (например, посредством функции со словарём, где ключом будет PID), но как это можно реализовать, если объявление классов требует наличие engine в самом начале? Наверное можно и определение классов перенести в функцию, но это какой-то кошмар... Какое здесь можно придумать нормальное решение?

Answer 1

Проблема в том, что при форке дочерний процесс копирует пул соединений и дескрипторы из родительского. Чтобы этого избежать есть разные варианты:

Вызывать engine.dispose() после форка во всех процессах. Но для этого нужно смешивать разные урони абстракции, мне не хотелось бы выносить работу с БД на уровень общей логики.
Поэтому я сделал проксирующий объект, который при обращении к движку отслеживает PID, и при его смене пересоздаёт пул подключений: engine.pool = engine.pool.recreate() Это позволяет не пробрасывать логику между разными уровнями абстракции и не сбрасывать пул подключений в родительском процессе. Надеюсь, при этом не возникает утечки памяти/дескрипторов ?

class PidLocalProxy:
    __proxied__ = (
        '_target', '_pid', '_on_change', '_validate_pid',
    )

    def __init__(self, target, on_change=None):
        self._target = target
        self._pid = os.getpid()
        self._on_change = on_change

    def __getattr__(self, attr):
        if attr not in PidLocalProxy.__proxied__:
            self._validate_pid()
            return getattr(self._target, attr)
        else:
            return super().__getattr__(attr)

    def __setattr__(self, attr, value):
        if attr not in PidLocalProxy.__proxied__:
            self._validate_pid()
            setattr(self._target, attr, value)
        else:
            super().__setattr__(attr, value)

    def _validate_pid(self):
        new_pid = os.getpid()
        if self._pid != new_pid:
            if self._on_change is not None:
                self._on_change(self._target)
                self._pid = new_pid
            else:
                raise ValueError(
                    f'{self!r} PID changed'
                    f' from {self._pid} to {new_pid}'
                )

    def __repr__(self):
        return f'PidLocalProxy({self._target !r})'


def refresh_engine(engine):
    engine.pool = engine.pool.recreate()

my_engine = PidLocalProxy(
    create_engine_by_info(my_config),
    on_change=refresh_engine,
)

Есть подобные рекомендации по работе с форком в официальной документации, но посредством прослушки событий: docs.sqlalchemy.org: Using Connection Pools with Multiprocessing or OS Fork. Однако у меня не получилось таким способом быстро достичь того же поведения, что и в варианте 2 (почему-то ломается подключение в родительском процессе).

БЛОГ НА HUSL

SQLAlchemy: automap_base и fork процесса

Ответы (1 шт):