Введение в модели SQLAlchemy

Перед тем как приступить к написанию наших первых запросов к БД, нам нужно настроить способ передачи сессии. Предпочтительным способом передачи сессии базы данных в функции обработки запроса представляет внедрение сессии бд в качестве зависимости. Поэтому создадим функцию get_db(), через которую объект сессии базы данных будет передаваться в функцию обработки. В папке app/backend cоздадим файл db_depends.py следующего содержания:

from app.backend.db import SessionLocal


async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Здесь сначала создаем объект сессии базы данных импортированный из файла db.py. Затем в конструкции try, finally с помощью оператора yield возвращаем созданный объект. Таким образом, данный объект будет внедрен в функцию обработки запроса. Выражение yield будет выполняться при получении каждого нового запроса.

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

После завершения операций с базой данных выполняется блок finally, в котором закрывается подключение к базе данных с помощью метода close()

Подключать этот код мы будем с помощью класса Depends, который мы изучали в этом разделе, в функцию передается результат функции get_db, то есть сессия базы данных, который передается параметру db. И через этот параметр мы сможем взаимодействовать с базой данных. Подобным образом сессия базы данных будет внедряться во все остальные функции-обработчики запросов.

Новый интерфейс запросов

В SQLAlchemy 2.0 появился новый интерфейс запросов. Если быть точным, эта функция была представлена в релизе SQLAlchemy 1.4 как способ помочь разработчикам перейти на версию 2.0.

До этого основной (но теперь уже устаревший) способ выполнения запросов в ORM SQLAlchemy заключался в использовании объекта Query, доступного из метода Session.query().

user = session.query(User).filter(User.username == 'Ilya').first()

В релизе SQLAlchemy 2.0 это теперь считается устаревшим способом выполнения запросов. Разработчики по-прежнему могут делать запросы этими методами, но в документации к SQLAlchemy такой подход уже называется «API запросов 1.x» или «устаревший API запросов».

Новый Query API имеет четкое разделение между самими запросами и средой выполнения, в которой они выполняются. Приведенный выше запрос для поиска пользователя по атрибуту username теперь можно записать вот так:

query = select(User).where(User.username == 'Ilya')

В этом примере запрос сохраняется в переменной query. При этом сейчас запрос еще не выполнен и даже пока не связан с сеансом. Для выполнения этого запроса его нужно передать в метод execute() объекта сеанса:

results = session.execute(query)

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

Метод all() позволяет вернуть list с объектом строки для каждой строки результата
Метод first() вернет первую строку результата
Метод one() вернет первую строку результата и вызовет исключение, если в ответе нет результата. Либо в объекте есть несколько одинаковых объектов, которые подходят под результат
Метод one_or_none() вернет первую строку результата, None — если результатов нет, или вызовет исключение, если есть более чем один подходящий к результату объект.

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

Метод scalars() возвращает ScalarResult объект с первым значением каждой строки результата. И перечисленные выше методы остаются доступны для этого нового объекта результата.
Метод scalar() возвращает первое значение первой строки результата.

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

user = session.scalar(query)

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