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 причинам:

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

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

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

from sqlalchemy.orm import sessionmaker
from database import engine

Session = sessionmaker(engine)

with Session() as session:
	#какие-то операции с БД

Тут мы импортировали engine из database.py, т.к. сессии нужно передать доступ к БД. Затем создали объект Session с помощью "Фабрики сессий" - sessionmaker и открыли сессию.

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

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

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

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

from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy import String

from database import engine

Session = sessionmaker(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, session) -> None:
	session.add(user)

    
with Session() as session:
    try:
        create_user(joe, session)
    except:
        session.rollback()
        raise
    else:
        session.commit()

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

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

Создали объект модели данных, который в таблице станет строкой (скаляром);
Написали функцию создания объекта в таблице, которая создаёт пользователя в БД
Открыли сессию, которая пытается создать объект в БД, если возникают ошибки - вызывается session.rollback, все изменения откатываются и сессия закрывается, если ошибок нет - вызывается session.commit, все изменения записываются в БД и сессия закрывается.

Получения объекта из БД

...
# +1 к списку импортов
from sqlalchemy import select 

# какой-то код, например тот, который писали выше

def get_by_name(name: str, session) -> list[UserBase]:
	statement = select(UserBase).where(UserBase.name == name)
	db_object = session.scalars(statement).one()
	return db_object


with Session() as session:
    print(get_by_name("Joe", session))

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

statement - наш запрос, select выбирает все объекты из таблицы "users";
.where - с английского звучит как "где", т.е. выбрать те объекты, где: UserBase.name(имя объекта из БД) равен нашему name, который мы передали через аргумент;
С помощью scalars мы получаем скаляры, то бишь строки из БД, по нашему запросу;
Возвращаем первый объект с помощью .one(), эта странная штука нам возвращает объект типа нашей модели данных, чтобы мы могли уже полноценно с ними работать (со скалярами мы мало чего сделаем). Также можно написать .all(), она вернёт все объекты , которые подходят по условию.

И что же будет? Мы получим что-то очень для нас не понятное (к примеру, "<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(new_object: UserBase, session) -> None:
	session.merge(new_object)

with Session() as session:
    try:
        joe = get_by_name("Joe", session)
        joe.name = "Bob"
        update(joe, session)
    except:
        session.rollback()
        raise
    else:
        session.commit()

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

Для начала мы можем получить объект из БД;
Изменить его атрибуты под свои нужды;
Передать его в функцию через аргумент;
Теперь сама функция: передаём новые значения из new_object в БД c помощью merge();
Подтверждаем изменения.

Как работает функция merge(): если в БД нет объекта с таким id, то она создаёт новый объект, если же есть - обновляет существующий.

Обратите внимание, для работы функции merge() нужно, чтобы в новом объекте значение атрибута id было равно значению этого же атрибута у объекта из БД.

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

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

def delete(name: str, session) -> UserBase:
	statement = select(UserBase).where(UserBase.name == name)
	db_object = session.scalars(statement).one()

	session.delete(db_object)
	return db_object

with Session() as session:
    try:
        delete("Joe", session)
    except:
        session.rollback()
        raise
    else:
        session.commit()

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

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

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