Библиотека SQLAlchemy: возможности и примеры использования

Расширенные примеры использования SQLAlchemy

Работа с отношениями: один ко многим и многие ко многим

Определим модели для блога: пользователь и статья (один ко многим) и теги (многие ко многим).

from sqlalchemy import ForeignKey, Table, Text
from sqlalchemy.orm import relationship

# вспомогательная таблица для many-to-many
post_tags = Table(
    'post_tags', Base.metadata,
    Column('post_id', Integer, ForeignKey('posts.id')),
    Column('tag_id', Integer, ForeignKey('tags.id'))
)

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    body = Column(Text)
    user_id = Column(Integer, ForeignKey('users.id'))
    author = relationship('User', back_populates='posts')
    tags = relationship('Tag', secondary=post_tags, back_populates='posts')

class Tag(Base):
    __tablename__ = 'tags'
    id = Column(Integer, primary_key=True)
    name = Column(String(30), unique=True)
    posts = relationship('Post', secondary=post_tags, back_populates='tags')

# добавляем relationship в User
User.posts = relationship('Post', order_by=Post.id, back_populates='author')

Base.metadata.create_all(engine)

Запрос с жадной загрузкой связанных объектов:

from sqlalchemy.orm import selectinload

with Session(engine) as session:
    stmt = select(Post).options(selectinload(Post.tags)).where(Post.id == 1)
    post = session.scalar(stmt)
    print(post.title, [tag.name for tag in post.tags])

Первая статья ['Python', 'SQLAlchemy']

Использование hybrid_property для вычисляемых полей

Гибридные свойства позволяют вычислять значения на уровне Python и SQL.

from sqlalchemy.ext.hybrid import hybrid_property

class User(Base):
    __tablename__ = 'users_with_hybrid'
    id = Column(Integer, primary_key=True)
    first_name = Column(String(30))
    last_name = Column(String(30))

    @hybrid_property
    def full_name(self):
        return f"{self.first_name} {self.last_name}"

    @full_name.expression
    def full_name(cls):
        # SQL выражение для фильтрации по полному имени
        from sqlalchemy import case
        return case(
            (cls.first_name.isnot(None), cls.first_name + ' ' + cls.last_name),
            else_=cls.last_name
        )

Base.metadata.create_all(engine)

Запрос с фильтром по выражению:

with Session(engine) as session:
    stmt = select(User).where(User.full_name == 'Иван Петров')
    user = session.scalar(stmt)
    print(user.full_name)

Иван Петров

Кастомные типы данных: JSON и Enum

Для баз данных, поддерживающих JSON, можно определить свой тип, наследующий JSON в SQLAlchemy.

from sqlalchemy import JSON, Enum
import enum

class StatusEnum(enum.Enum):
    active = 'active'
    inactive = 'inactive'

class Config(Base):
    __tablename__ = 'configs'
    id = Column(Integer, primary_key=True)
    data = Column(JSON, nullable=True)
    status = Column(Enum(StatusEnum), default=StatusEnum.active)

Base.metadata.create_all(engine)

with Session(engine) as session:
    config = Config(data={"theme": "dark", "lang": "ru"}, status=StatusEnum.active)
    session.add(config)
    session.commit()
    # чтение
    stmt = select(Config).where(Config.data['theme'].as_string() == 'dark')
    result = session.scalars(stmt).first()
    print(result.data['lang'])

ru

Сложные запросы: подзапросы и CTE

Пример с использованием оконной функции для нумерации строк в группе:

from sqlalchemy import func, and_

with Session(engine) as session:
    subq = session.query(
        Post.id,
        Post.title,
        func.row_number().over(
            order_by=Post.id,
            partition_by=Post.user_id
        ).label('rn')
    ).subquery()
    stmt = select(subq).where(subq.c.rn == 1)
    result = session.execute(stmt).fetchall()
    for row in result:
        print(row.title)

Вывод: первая публикация каждого пользователя (по id).

Bulk-операции для массовой вставки

Метод insert().returning() в Core или bulk_insert_mappings в ORM.

users_data = [
    {"name": "Мария", "email": "maria@example.com"},
    {"name": "Сергей", "email": "sergey@example.com"},
]
with Session(engine) as session:
    session.bulk_insert_mappings(User, users_data, return_defaults=True)
    session.commit()
    for u in users_data:
        print(u['id'], u['name'])

2 Мария
3 Сергей

Асинхронная вставка с gather

Использование aiohttp и SQLAlchemy async для параллельной вставки.

import asyncio
from sqlalchemy.ext.asyncio import AsyncSession

async def insert_users():
    async with AsyncSessionLocal() as session:
        async with session.begin():
            users = [User(name=f'User {i}') for i in range(5)]
            session.add_all(users)

async def main():
    await asyncio.gather(insert_users(), insert_users())

asyncio.run(main())