Пример 1: Настройка окружения
Полный скрипт для запуска всех задач практикума в SQLite in-memory базе:
# practice_setup.py
from sqlalchemy import create_engine, func, ForeignKey, String, Integer
from sqlalchemy.orm import (
DeclarativeBase, Mapped, mapped_column,
relationship, Session
)
from typing import List, Optional
# --- Базовый класс ---
class Base(DeclarativeBase):
pass
# --- Модели ---
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
age: Mapped[int] = mapped_column(Integer)
addresses: Mapped[List["Address"]] = relationship(
"Address",
back_populates="user",
cascade="all, delete-orphan"
)
def __repr__(self) -> str:
return f"User(id={self.id}, name={self.name!r}, age={self.age})"
class Address(Base):
__tablename__ = 'addresses'
id: Mapped[int] = mapped_column(primary_key=True)
description: Mapped[Optional[str]] = mapped_column(String(255))
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
user: Mapped["User"] = relationship("User", back_populates="addresses")
def __repr__(self) -> str:
return f"Address(id={self.id}, description={self.description!r})"
# --- Создание таблиц ---
engine = create_engine('sqlite:///:memory:', echo=False)
Base.metadata.create_all(engine)
Пример 2: Заполнение тестовыми данными
with Session(engine) as session:
users_data = [
User(name="Alice", age=25),
User(name="Bob", age=30),
User(name="Charlie", age=22),
User(name="Diana", age=35),
User(name="Eve", age=28),
User(name="Frank", age=22), # одинаковый возраст с Charlie
]
session.add_all(users_data)
session.flush() # получить id без коммита
addresses_data = [
Address(description="Berlin", user=users_data[0]), # Alice
Address(description="Paris", user=users_data[1]), # Bob
Address(description="Berlin", user=users_data[2]), # Charlie
Address(description="London", user=users_data[3]), # Diana
# Eve и Frank — без адресов
]
session.add_all(addresses_data)
session.commit()
print("Тестовые данные загружены")
Пример 3: Типовые запросы каждого блока
Блок 1 — поиск и фильтрация
with Session(engine) as session:
# Найти пользователя по имени
user = session.query(User).filter(User.name == "Alice").first()
print(user) # User(id=1, name='Alice', age=25)
# Все старше 20
older = session.query(User.name).filter(User.age > 20).all()
print([u.name for u in older]) # ['Alice', 'Bob', 'Charlie', ...]
# Топ-4 по алфавиту
top4 = session.query(User.name).order_by(User.name).limit(4).all()
print([u.name for u in top4]) # ['Alice', 'Bob', 'Charlie', 'Diana']
Блок 2 — агрегации
with Session(engine) as session:
avg_age = session.query(func.avg(User.age)).scalar()
print(f"Средний возраст: {avg_age:.1f}") # 27.0
max_age = session.query(func.max(User.age)).scalar()
min_age = session.query(func.min(User.age)).scalar()
print(f"Макс: {max_age}, Мин: {min_age}") # 35, 22
# Группировка по возрасту
groups = session.query(User.age, func.count(User.id)).group_by(User.age).all()
for age, cnt in groups:
print(f"Возраст {age}: {cnt} чел.")
# Возраст 22: 2 чел. (Charlie + Frank)
# Возраст 25: 1 чел. ...
Блок 3 — JOIN
with Session(engine) as session:
# INNER JOIN: пользователи с адресами
rows = (
session.query(User.name, Address.description)
.join(User.addresses)
.all()
)
for name, addr in rows:
print(f"{name} -> {addr}")
# Alice -> Berlin
# Bob -> Paris ...
# LEFT JOIN: пользователи БЕЗ адресов
no_addr = (
session.query(User.name)
.outerjoin(User.addresses)
.filter(Address.id.is_(None))
.all()
)
print([u.name for u in no_addr]) # ['Eve', 'Frank']