Данная статья представляет собой ознакомление с базовым синтаксисом 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(new_object: UserBase) -> None:
	with Session(engine) as session:
		statement = select(UserBase).where(UserBase.name == new_object.name)
		db_object = session.scalars(statement).one()

		# небольшая плюшка
		for key, value in new_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 я передал, остальное уже отдельно можно подыскать, в зависимости от ваших задач.

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