FastAPI 3. Подключение к SQLAlchemy и генератор сессий

В предыдущих постах, мы с вами инициализировали проект и создали первоначальную конфигурацию.
Следующим шагом в разработке нам необходимо создать точку подключения к SQLAlchemy. Это поможет эффективно управлять сессиями базы данных и запросами. В процессе мы создадим класс, который будет содержать асинхронный движок базы данных, фабрику сессий и применим концепцию scoped session (сессии с ограниченной областью видимости).

Зачем нужен такой подход: асинхронное взаимодействие с базой данных

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

Асинхронные операции: скорость и отзывчивость

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

Асинхронный движок и сессии

Асинхронный движок базы данных, создаваемый через create_async_engine, позволяет нам отправлять запросы к базе данных в асинхронном режиме. Сессии, созданные через async_sessionmaker, управляют подключением к базе данных для каждой операции. Использование сессий позволяет:

1. Открывать отдельное подключение для каждого запроса.
2. Гарантировать, что сессия закрывается корректно после завершения работы (в нашем случае с помощью метода scoped_session_dependency).
3. Избежать возможных проблем с доступом к базе данных, таких как конкуренция запросов или утечка подключений.

Scoped session (сессии с ограниченной областью видимости)

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

Когда это особенно полезно?

Такой подход применяется в асинхронных веб-приложениях, которые обслуживают множество пользователей одновременно. Например, в API, которые принимают множество запросов от клиентов, такие как REST или GraphQL-сервисы. Асинхронность помогает обрабатывать тысячи запросов одновременно, не блокируя приложение на время выполнения операций с базой данных.

Создание файла database.py

Для начала, в пакете core, там, где у нас уже существует файл settings.py, создадим новый файл — database.py. Этот файл будет отвечать за настройку базы данных и создание сессий.

Класс Database

В файле database.py создадим класс Database. Этот класс будет содержать два ключевых метода: конструктор класса и метод для работы с сессиями.

Конструктор класса

Первым методом будет конструктор класса — так называемый dunder-метод __init__. Он принимает два аргумента:

1. url — строка с адресом для подключения к базе данных. Этот адрес у нас уже прописан в классе Settings.
2. echo — булево значение, определяющее режим "отладки". Если echo=True, будут выводиться SQL-запросы в консоль, что полезно для разработки.

Внутри конструктора создаются два основных поля:

1. self.engine — здесь создаётся асинхронный движок базы данных с помощью функции create_async_engine из SQLAlchemy, которой передаются аргументы url и echo.
2. self.session_factory — создаётся асинхронная фабрика сессий с помощью async_sessionmaker. Фабрика используется для создания отдельных сессий базы данных. Здесь мы устанавливаем три параметра (autoflush, autocommit, expire_on_commit) в False, чтобы более точно контролировать работу с сессиями.

Код конструктора:

def __init__(self, url: str, echo: bool):  
    self.engine = create_async_engine(url=url, echo=echo)  
    self.session_factory = async_sessionmaker(  
        bind=self.engine, autoflush=False, autocommit=False, expire_on_commit=False  
    )

Асинхронный метод scoped_session_dependency

Вторым важным методом будет асинхронный метод scoped_session_dependency. Это асинхронный генератор, который возвращает объект класса AsyncGenerator.

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

Логика работы метода:

1. В обработчике запросов вызывается метод scoped_session_dependency().
2. Метод получает новую сессию из фабрики self.session_factory.
3. Дойдя до оператора yield, метод возвращает сессию туда, где был вызван, и приостанавливает свою работу.
4. После завершения работы обработчика запросов метод возобновляет выполнение.
5. Выполняется закрытие сессии через await session.close().

Код метода:

async def scoped_session_dependency(self) -> AsyncGenerator:  
    async with self.session_factory() as session:  
        yield session  
        await session.close()

Экземпляр класса Database

После того, как мы определили класс Database, нам нужно создать его экземпляр. Для этого под классом создаём переменную database. При её инициализации вызываем класс Database и передаём два обязательных аргумента в конструктор: url и echo. Оба значения мы получаем из переменной settings, которая находится в файле settings.py.

Важное замечание

Обратите внимание: в поле db_url хранится не строка с адресом подключения, а объект класса PostgresDsn. Чтобы получить строку для подключения, необходимо вызвать метод .unicode_string().

Код инициализации:

database = Database(url=settings.db_settings.db_url.unicode_string(), echo=settings.db_settings.db_echo)

Полный код файла

from typing import AsyncGenerator  

from sqlalchemy.ext.asyncio import (  
    create_async_engine,  
    async_sessionmaker,  
)  

from lkeep.core.settings import settings  


class Database:  
    def __init__(self, url: str, echo: bool):  
        self.engine = create_async_engine(url=url, echo=echo)  
        self.session_factory = async_sessionmaker(  
            bind=self.engine, autoflush=False, autocommit=False, expire_on_commit=False  
        )  

    async def scoped_session_dependency(self) -> AsyncGenerator:  
        async with self.session_factory() as session:  
            yield session  
            await session.close()  


database = Database(url=settings.db_settings.db_url.unicode_string(), echo=settings.db_settings.db_echo)

Заключение

Созданный нами класс Database обеспечивает эффективное управление сессиями базы данных. С его помощью мы можем безопасно и быстро работать с базой данных: открывать сессии для выполнения запросов и автоматически закрывать их после завершения работы. В дальнейшем этот класс будет применён при написании обработчиков запросов (представлений), обеспечивая корректное выполнение операций с базой данных.