SQLAlchemy 2.0 для новичков

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

Предполагается, что вы знакомы с базовым синтаксисом языка Python и, возможно, новичок в программировании.

Установка

$ pip install SQLAlchemy

Создание модели данных

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

from sqlalchemy.orm import DeclarativeBase  class Base(DeclarativeBase): pass 

Класс Base станет нашей отправной точкой в создании моделей, обычно модели в SQLAlchemy называют так, что в конце названия красуется «Base»: CarBase, HumanBase, ProductBase и т.п. Это улучшит читаемость кода как для Вас, так и для тех, кому придётся его читать.

Теперь мы можем создавать наши модели данных:

from sqlalchemy.orm import DeclarativeBase from sqlalchemy.orm import Mapped from sqlalchemy.orm import mapped_column from sqlalchemy import String  class Base(DeclarativeBase): pass  class UserBase(Base): __tablename__ = "users"  id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(30)) 

Разберём код, __tablename__ — название таблицы в базе данных.
Mapped[type] — транслирует тип данных Python в тип данных SQL (К примеру int в INTEGER, str в VARCHAR).
mapped_colum — позволяет задать валидацию данных (к примеру, максимальная длина строки String(30)), определить первичный ключ, т.е. id или uuid (primary_key=True), который будет определяться автоматически, при занесении в таблицу, а также определить взаимоотношения, подробнее о них ниже.

Что такое relationships

Relationship позволяет создать связи между колонками как внутри одной таблицы, так и между несколькими таблицами.

Допустим есть у нас люди и автомобили, у некоторых людей есть авто, у некоторых нет, как нам реализовать такие связи? Да очень просто на самом деле

from sqlalchemy import ForeignKey # предыдущие импорты ...  class Human(Base):     __tablename__ = "humans"      id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String())  class Car(Base):     __tablename__ = "cars"      id: Mapped[id] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String()) owner_id: Mapped[int] = mapped_column(ForeignKey("Human.id"))  joe = Human(name="Joe") vaz_1111 = Car(name="Ока", owner_id=1) 

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

Необязательные поля

Название говорит само за себя, так что приступим

from typing import Optional ...  class BomBom(Base):     __tablename__ = "bomboms"      id: Mapped[int] = mapped_column(primary_key=True) bom_bom: Mapped[Optional[str]] = mapped_column(String())  bom_one = BomBom() bom_two = BomBom(bom_bom="Бом-Бом") 

Пусть будет BomBom.
В чём суть: мы оборачиваем тип данных колонны в Optional[], благодаря чему значение становится необязательным и может быть равно None

Создание и подключение БД

Подключаем БД:

from sqlalchemy import create_engine engine = create_engine("sqlite:///(путь к БД)", echo=True) 

Нетрудно догадаться что делает этот код, разве что echo=True может создать вопрос, на который есть ответ: этот атрибут включает логирование событий БД (например, занесение данных в таблицу). Перейдём к созданию БД.

from sqlalchemy import create_engine from models import Base  DB_URL = 'sqlite:///db/database.db' engine = create_engine(DB_URL, echo=True)  def create_db_and_tables() -> None: Base.metadata.create_all(engine) 

Base.metadata.create_all(engine) — создаёт таблицы, на основе объявленных моделей, здесь она ничего не создаст по 2 причинам:

1. Функция не вызывается 🙂

2. Не объявлена ни одна модель, в этом файле, т.е. в начало файла нужно добавить импорт нашей модели: from models import UserBase. После того как мы вызовем функцию create_db_and_tables() у нас создастся БД database.db, в которой будет таблица «users».

Создание сессии в БД

Чтобы мы могли взаимодействовать с БД нам нужно открыть сессию, т.е. создать объект сессии, для этого мы будем использовать контекстный менеджер with (если не знаете как он работает, хабр вам в помощь).

from sqlalchemy.orm import Session from database import engine  with Session(engine) as session: #какие-то операции с БД 

Тут мы импортировали engine из database.py, т.к. сессии нужно передать доступ к БД, и открыли сессию.

Взаимодействие с БД

Переходим к самому интересному: организуем CRUD-функции (create, read, update, delete). Программы, которые напрямую работают с БД и выполняют выше приведённые функции называются репозиториями.

Создание объекта

Для начала функция создания объекта:

from sqlalchemy.orm import Session from sqlalchemy.orm import Mapped from sqlalchemy.orm import mapped_column from sqlalchemy import String  from database import engine  # перенесём сюда модель для наглядности class UserBase(Base): __tablename__ = "users"  id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(30)) email: Mapped[str] = mapped_column(String(100))  joe = UserBase(name="Joe", email="joe@example.com")  def create_user(user: UserBase) -> None: with Session(engine) as session: session.add(user) session.commit() session.refresh(user)  create_user(joe) 

p.s. вообще можно создать конструктор объектов, но для примера опустим этот момент.

Что мы наделали, собственно:

1. Создали объект модели данных, который в таблице станет строкой (скаляром);

2. Написали функцию создания объекта в таблице:

   1. Открыли сессию;

   2. Добавили пользователя и подтвердили операцию(session.add(), session.commit());

   3. Обновили наши знания о данных в таблице(session.refresh()).

Получение объекта

... # +1 к списку импортов from sqlalchemy import select   # какой-то код, например тот, который писали выше  def get_by_name(name: str) -> list[UserBase]: with Session(engine) as session: statement = select(UserBase).where(UserBase.name == name) objects = session.scalars(statement).all() return objects  print(get_by_name("Joe")) 

Так, по порядку:

1. statement — наш запрос, select выбирает все объекты из таблицы «users»;

2. .where — с английского звучит как «где», т.е. выбрать те объекты, где: UserBase.name(имя объекта из БД) равен нашему name, который мы передали через аргумент;

3. С помощью scalars мы получаем скаляры, то бишь строки из БД, по нашему запросу;

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

И что же будет? Мы получим что-то очень для нас не понятное (к примеру, "<app.models.UserBase object at 0x7dbc3b8c41a0>"), потому что нужно было добавить __repr__() в модель данных (Это нужно только для вывода в консоль).

class UserBase(Base): __tablename__ = "users"  id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(30)) email: Mapped[str] = mapped_column(String(100))  def __repr__(self) -> str: return f"UserBase(id={self.id}, name={self.name}, email={self.email})" 

Теперь мы можем выводить в консоль объекты типа UserBase:

[UserBase(id=1, name=Joe, email=joe@examle.com)]

Обновление объекта

def update(object: UserBase) -> None: with Session(engine) as session: statement = select(UserBase).where(UserBase.name == object.name) db_object = session.scalars(statement).one()  # небольшая плюшка for key, value in object.__dict__.items(): if (key != "id" and key != "_sa_instance_state" and value is not None): setattr(db_object, key, value)  session.commit() session.refresh(db_object)  update(new_joe) 

Разберём написанный код:

1. Для начала мы можем получить объект из БД;

2. Изменить его атрибуты под свои нужды;

3. Передать его в функцию через аргумент;

4. Теперь сама функция: также получаем объект из БД;

5. Далее присваиваем все атрибуты нового объекта старому, кроме id, которого у нашего объекта нет, т.к. он не был в БД и у него просто нет значения этого атрибута, и _sa_instance_state, это атрибут связывающий объект с текущей сессией, его нам менять не нужно ни в коем случае.

.__dict__.items() возвращает нам все атрибуты объекта в виде словаря, setattr() задаёт значение атрибута.

Удаление объекта

Тут всё просто, получаем объект, как раньше, и удаляем его одной простой командой.

def delete(name: str) -> UserBase: with Session(engine) as session: statement = select(UserBase).where(UserBase.name == name) object = session.scalars(statement).one()  session.delete(object) session.commit() return object  

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

Если наберём два с половиной лайка выложу статью о том, как можно ко всему этому прикрутить дженерики, чтобы сделать универсальный репозиторий для работы с любыми моделями данных.