💻 Примеры кода блока SQLAlchemy

Пример 1. Создание модели и таблицы

Полный пример от определения модели до создания таблицы в базе:

from decimal import Decimal
from sqlalchemy import Boolean, ForeignKey, Numeric, String, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship

engine = create_engine('sqlite:///example.db')

class Base(DeclarativeBase):
    pass

class Category(Base):
    __tablename__ = 'categories'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    description: Mapped[str | None] = mapped_column(String(255))
    products: Mapped[list["Product"]] = relationship(back_populates="category")

class Product(Base):
    __tablename__ = 'products'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    price: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    in_stock: Mapped[bool] = mapped_column(Boolean, default=True)
    category_id: Mapped[int] = mapped_column(ForeignKey('categories.id'))
    category: Mapped["Category"] = relationship(back_populates="products")

Base.metadata.create_all(engine)

Пример 2. CRUD: Create

with Session(engine) as session:
    categories = [
        Category(name="Электроника", description="Гаджеты и устройства."),
        Category(name="Книги", description="Печатные и электронные книги."),
        Category(name="Одежда", description="Одежда для мужчин и женщин."),
    ]
    session.add_all(categories)
    session.flush()

    products = [
        Product(name="Смартфон", price=299.99, in_stock=True, category=categories[0]),
        Product(name="Ноутбук", price=499.99, in_stock=True, category=categories[0]),
        Product(name="Роман", price=15.99, in_stock=True, category=categories[1]),
        Product(name="Джинсы", price=40.50, in_stock=False, category=categories[2]),
        Product(name="Футболка", price=20.00, in_stock=True, category=categories[2]),
    ]
    session.add_all(products)
    session.commit()

Пример 3. CRUD: Read с фильтрацией

from sqlalchemy import select

with Session(engine) as session:
    for category in session.scalars(select(Category)).all():
        print(f"Категория: {category.name}")
        for p in category.products:
            print(f"  - {p.name}: {p.price} ({'есть' if p.in_stock else 'нет'})")

    expensive_in_stock = session.scalars(
        select(Product).where(Product.in_stock.is_(True), Product.price > 100)
    ).all()
    for p in expensive_in_stock:
        print(p.name, p.price)

Пример 4. CRUD: Update и Delete

with Session(engine) as session:
    product = session.scalars(
        select(Product).where(Product.name == "Смартфон")
    ).first()
    if product:
        product.price = 349.99
        session.commit()

with Session(engine) as session:
    product = session.scalars(
        select(Product).where(Product.name == "Футболка")
    ).first()
    if product:
        session.delete(product)
        session.commit()

Пример 5. Агрегации и группировка

from sqlalchemy import func, select

with Session(engine) as session:
    result = session.execute(
        select(Category.name, func.count(Product.id).label('total'))
        .join(Product)
        .group_by(Category.name)
    ).all()

    for row in result:
        print(f"{row.name}: {row.total} продуктов")

    cats = session.execute(
        select(Category.name)
        .join(Product)
        .group_by(Category.name)
        .having(func.count(Product.id) > 1)
    ).all()
    print("Категории с 2+ продуктами:", [c.name for c in cats])

Пример 6. Логические условия и сортировка

from sqlalchemy import desc, or_, select

with Session(engine) as session:
    session.scalars(
        select(Product).where(or_(Product.in_stock.is_(True), Product.price < 20))
    ).all()

    session.scalars(select(Product).order_by(desc(Product.price))).all()

    session.scalars(
        select(Product).order_by(desc(Product.price), Product.name)
    ).all()

Пример 7. JOIN

with Session(engine) as session:
    products_with_category = session.scalars(select(Product).join(Category)).all()

    all_products = session.scalars(select(Product).outerjoin(Category)).all()

    rows = session.execute(
        select(Product.name, Category.name).join(Category)
    ).all()

Пример 8. Подзапрос

from sqlalchemy import func, select

avg_price_sub = select(func.avg(Product.price).label('avg_price')).subquery()

with Session(engine) as session:
    expensive = session.scalars(
        select(Product).where(Product.price > avg_price_sub.c.avg_price)
    ).all()
    for p in expensive:
        print(p.name, p.price)