ORM на примере SQLAlchemy

from sqlalchemy import create_engine, event, ForeignKey, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    name: Mapped[str | None]                                   # необязательное поле (NULL)
    posts: Mapped[list["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")

    def __repr__(self):
        return f"User(id={self.id}, email={self.email!r})"

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    likes: Mapped[int] = mapped_column(default=0)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped[User] = relationship(back_populates="posts")

engine = create_engine("sqlite://")              # с echo=True SQLAlchemy сам логирует весь SQL
Base.metadata.create_all(engine)                 # CREATE TABLE (в проекте — миграции Alembic)

@event.listens_for(engine, "before_cursor_execute")
def show_sql(conn, cursor, statement, parameters, context, executemany):
    print("SQL>", " ".join(statement.split()), parameters)

with Session(engine) as session:
    anna = User(email="anna@example.com", name="Анна")
    anna.posts.append(Post(title="Привет, SQLAlchemy"))
    session.add(anna)
    session.commit()                             # INSERT в одной транзакции
    print(anna.id, anna.posts[0].author)

from sqlalchemy import create_engine, ForeignKey, String, select, func, or_
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    city: Mapped[str]
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    likes: Mapped[int] = mapped_column(default=0)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped[User] = relationship(back_populates="posts")

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
session = Session(engine)
anna, boris, vera = User(name="Анна", city="Казань"), User(name="Борис", city="Москва"), User(name="Вера", city="Москва")
anna.posts = [Post(title="Генераторы", likes=40), Post(title="asyncio", likes=95)]
boris.posts = [Post(title="Docker за час", likes=70)]
session.add_all([anna, boris, vera])
session.commit()

# объекты целиком
for user in session.scalars(select(User).where(User.city == "Москва").order_by(User.name)):
    print(user.name)

# один объект
print(session.get(User, 1).name)                                            # по первичному ключу
print(session.scalars(select(User).where(User.name == "Вера")).one().city)   # .one(), .first(), .one_or_none()

# отдельные колонки, условия, LIMIT
stmt = select(Post.title, Post.likes).where(or_(Post.likes > 80, Post.title.like("D%"))).order_by(Post.likes.desc()).limit(5)
print(session.execute(stmt).all())

# JOIN и агрегаты
stmt = (
    select(User.name, func.count(Post.id).label("posts"), func.coalesce(func.sum(Post.likes), 0))
    .outerjoin(User.posts)
    .group_by(User.id)
    .order_by(User.name)
)
print(stmt)          # можно посмотреть сгенерированный SQL
print(session.execute(stmt).all())

from sqlalchemy import create_engine, select, update, delete
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    price: Mapped[int]
    stock: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add_all([Product(title="Кружка", price=500, stock=3), Product(title="Футболка", price=1500),
                     Product(title="Блокнот", price=300, stock=10)])
    session.commit()

    mug = session.scalars(select(Product).where(Product.title == "Кружка")).one()
    mug.price = 450                      # просто меняем атрибут — сессия отследит
    session.commit()                     # UPDATE products SET price=450 WHERE id=1

    session.execute(update(Product).where(Product.stock == 0).values(price=Product.price * 0.9))   # массовое обновление
    session.execute(delete(Product).where(Product.price < 400))                                   # массовое удаление
    session.commit()

    print([(p.title, p.price) for p in session.scalars(select(Product).order_by(Product.id))])

    try:
        with session.begin_nested():     # точка сохранения (SAVEPOINT)
            session.add(Product(title="Ошибка", price=None))
            session.flush()              # отправить в базу сейчас → IntegrityError
    except Exception as e:
        print("откат:", type(e).__name__)
    print(session.scalars(select(Product.title)).all())

from sqlalchemy import create_engine, ForeignKey, select, event
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session, selectinload, joinedload

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped[User] = relationship(back_populates="posts")

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
with Session(engine) as s:
    s.add_all([User(name=f"user{i}", posts=[Post(title=f"p{i}-{j}") for j in range(3)]) for i in range(20)])
    s.commit()

queries = []
event.listen(engine, "before_cursor_execute", lambda *args: queries.append(args[2]))

def count_queries(stmt, label):
    queries.clear()
    with Session(engine) as s:
        total = sum(len(u.posts) for u in s.scalars(stmt).unique())
    print(f"{label:14} постов: {total}, SQL-запросов: {len(queries)}")

count_queries(select(User), "lazy")
count_queries(select(User).options(selectinload(User.posts)), "selectinload")
count_queries(select(User).options(joinedload(User.posts)), "joinedload")
print(queries[-1][:120], "...")

from sqlalchemy import create_engine, Table, Column, ForeignKey, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase):
    pass

student_course = Table(                       # промежуточная таблица
    "student_course", Base.metadata,
    Column("student_id", ForeignKey("students.id"), primary_key=True),
    Column("course_id", ForeignKey("courses.id"), primary_key=True),
)

class Student(Base):
    __tablename__ = "students"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    courses: Mapped[list["Course"]] = relationship(secondary=student_course, back_populates="students")

class Course(Base):
    __tablename__ = "courses"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    students: Mapped[list[Student]] = relationship(secondary=student_course, back_populates="courses")

engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
with Session(engine) as s:
    python, sql = Course(title="Python"), Course(title="SQL")
    s.add_all([Student(name="Аня", courses=[python, sql]), Student(name="Борис", courses=[python])])
    s.commit()
    print([st.name for st in python.students])
    print(s.scalars(select(Student.name).join(Student.courses).where(Course.title == "SQL")).all())

from sqlalchemy import create_engine, text

engine = create_engine("sqlite://")
with engine.begin() as conn:                 # транзакция: commit при выходе, rollback при ошибке
    conn.execute(text("CREATE TABLE t (x INTEGER)"))
    conn.execute(text("INSERT INTO t VALUES (:x)"), [{"x": i} for i in range(5)])
    print(conn.execute(text("SELECT sum(x) FROM t WHERE x > :min"), {"min": 1}).scalar())