Ленивое отражение таблиц в SQLAlchemy ORM

SQLAlchemy существует на рынке уже долгое время и является одним из лучших ORM, доступных на сегодняшний день. При работе над бэкенд-фреймворками, такими как Flask или FastAPI, мы обычно сталкиваемся с этим ORM.

Есть два подхода, мы можем использовать SQLAlchemy:

• Создание схемы, таблиц вручную с использованием объектов declarative_base и их перенос.
• Отражение существующих объектов в базе данных с помощью metadata.

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

Идея заключается в том, чтобы лениво отражать таблицы и представления в зависимости от требований, а не загружать все сразу. Нам ведь не нужны все таблицы сразу, не так ли?

Поскольку API будут запрашивать или выполнять CRUD-операции только к подмножеству таблиц, у нас есть возможность пропустить загрузку других таблиц, но при этом сохранить постоянство уже отраженных таблиц.

Для этого я создал lazy обертку, которая отражает таблицы в требованиях только один раз и сохраняет их, чтобы вы могли использовать те же объекты в дальнейшем.

class LazyDBProp(object):
    """This descriptor returns sqlalchemy
    Table class which can be used to query
    table from the schema
    """

    def __init__(self) -> None:
        self._table = None
        self._name = None

    def __set_name__(self, _, name):
        self._name = name

    def __set__(self, instance, value):
        if isinstance(value, (CustomTable, Table)):
            self._table = value

    def __get__(self, instance, _):
        if self._table is None:
            self._table = CustomTable(
                self._name, instance.metadata, autoload=True)
        return self._table

Этот класс использует дескрипторы под капотом для сохранения объектов table или view. Я также создал обертку для создания динамического класса для хранения этих объектов таблицы на основе дескрипторов.

def get_lazy_class(engine: Engine) -> object:
    """
    Function to create Lazy class for pulling table object
    using SQLalchemy metadata
    """

    def __init__(self, engine: Engine):
        self.metadata = MetaData(engine)
        self.engine = engine

    def __getattr__(self, attr):
        if attr not in self.__dict__:
            obj = self.__patch(attr)
        return obj.__get__(self, type(self))

    def __patch(self, attribute):
        obj = LazyDBProp()
        obj.__set_name__(self, attribute)
        setattr(type(self), attribute, obj)
        return obj

    # naming classes uniquely for different schema's
    # to avoid cross referencing
    LazyClass = type(f"LazyClass_{engine.url.database}", (), {})
    LazyClass.__init__ = __init__
    LazyClass.__getattr__ = __getattr__
    LazyClass.__patch = __patch
    return LazyClass(engine)

Приведенный выше класс можно просто использовать как показано ниже:

from lazy_alchemy import get_lazy_class
from sqlalchemy import create_engine

db_engine = create_engine(DB_CONNECT_STRING)
lazy_db = get_lazy_class(db_engine)

db_model = lazy_db.my_db_table_foo
query = session.query(db_model).filter(db_model.foo == "bar").all()

После отражения на эти объекты можно ссылаться многократно. Отражение только необходимых объектов повышает производительность приложения с минимальными накладными расходами.

Это позволило мне сократить время загрузки приложения с более чем минуты до пары секунд :).

Если вы хотите реализовать вышеописанное в своем проекте, вы можете просто использовать мой pypi пакет Lazy Alchemy.

Я буду рад услышать ваши мнения и альтернативы этому подходу.

Спасибо, что прочитали эту статью, надеюсь, вы нашли для себя полезные советы.