Hybrid properties, association proxies, computed columns
Продвинутые паттерны SQLAlchemy: hybrid properties для атрибутов работающих на Python и SQL, association proxies для удобного доступа к связям, computed columns и другие техники.
Hybrid property работает и на уровне экземпляра (Python), и на уровне класса (SQL выражение).
from sqlalchemy.ext.hybrid import hybrid_property
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
first_name: Mapped[str] = mapped_column(String(100))
last_name: Mapped[str] = mapped_column(String(100))
@hybrid_property
def full_name(self) -> str:
"""Python уровень: работает с экземпляром."""
return f"{self.first_name} {self.last_name}"
@full_name.expression
def full_name(cls):
"""SQL уровень: работает с классом для запросов."""
from sqlalchemy import func
return func.concat(cls.first_name, ' ', cls.last_name)
# Использование
# Python уровень
user = session.get(User, 1)
print(user.full_name) # "John Doe"
# SQL уровень
stmt = select(User).where(User.full_name == 'John Doe')
users = session.execute(stmt).scalars().all()
# Агрегации
stmt = select(func.count(User.id)).where(User.full_name.like('%Smith%'))from sqlalchemy.ext.hybrid import hybrid_property
from datetime import datetime, date
class Article(Base):
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(primary_key=True)
published_at: Mapped[datetime] = mapped_column(DateTime)
view_count: Mapped[int] = mapped_column(Integer, default=0)
@hybrid_property
def is_recent(self) -> bool:
"""Проверка на уровне Python."""
week_ago = datetime.utcnow() - timedelta(days=7)
return self.published_at > week_ago
@is_recent.expression
def is_recent(cls):
"""Проверка на уровне SQL."""
from sqlalchemy import func, text
return cls.published_at > (func.now() - text("INTERVAL '7 days'"))
@hybrid_property
def popularity(self) -> str:
"""Категория популярности."""
if self.view_count < 100:
return 'low'
elif self.view_count < 1000:
return 'medium'
else:
return 'high'
@popularity.expression
def popularity(cls):
"""SQL версия с CASE."""
from sqlalchemy import case
return case(
(cls.view_count < 100, 'low'),
(cls.view_count < 1000, 'medium'),
else_='high'
)
# Использование в запросах
stmt = select(Article).where(Article.is_recent == True)
recent_articles = session.execute(stmt).scalars().all()
stmt = select(Article).where(Article.popularity == 'high')
popular = session.execute(stmt).scalars().all()from sqlalchemy.ext.hybrid import hybrid_method
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
posts: Mapped[list["Post"]] = relationship()
@hybrid_method
def has_more_posts_than(self, threshold: int) -> bool:
"""Python версия."""
return len(self.posts) > threshold
@has_more_posts_than.expression
def has_more_posts_than(cls, threshold: int):
"""SQL версия."""
from sqlalchemy import select, func
return (
select(func.count(Post.id))
.where(Post.user_id == cls.id)
.correlate(cls)
.scalar_subquery()
) > threshold
# Использование
# Python
user = session.get(User, 1)
if user.has_more_posts_than(5):
print("Active user")
# SQL
stmt = select(User).where(User.has_more_posts_than(5))
active_users = session.execute(stmt).scalars().all()Association proxy даёт удобный доступ к атрибутам связанных объектов.
from sqlalchemy.ext.associationproxy import association_proxy
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
keywords: Mapped[list["Keyword"]] = relationship(
back_populates="user",
cascade="all, delete-orphan"
)
# Proxy для доступа к значениям ключевых слов
keyword_values = association_proxy('keywords', 'keyword')
def __init__(self, keyword_values=None):
if keyword_values:
for kw in keyword_values:
self.keywords.append(Keyword(keyword=kw))
class Keyword(Base):
__tablename__ = 'keywords'
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'), primary_key=True)
keyword: Mapped[str] = mapped_column(String(50), primary_key=True)
user: Mapped["User"] = relationship(back_populates="keywords")
# Использование
user = User(keyword_values=['python', 'sqlalchemy', 'fastapi'])
session.add(user)
session.commit()
# Доступ через proxy
print(user.keyword_values) # ['python', 'sqlalchemy', 'fastapi']
# Добавление через proxy
user.keyword_values.append('alembic')
session.commit()
# Проверка вхождения
if 'python' in user.keyword_values:
print("Has python keyword")
# Запросы
from sqlalchemy import select
stmt = select(User).where(User.keyword_values.any('python'))from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy import Column, ForeignKey, String, Integer, DateTime
from datetime import datetime
# Ассоциативная таблица с дополнительными полями
class UserTag(Base):
__tablename__ = 'user_tags'
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'), primary_key=True)
tag_id: Mapped[int] = mapped_column(ForeignKey('tags.id'), primary_key=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
score: Mapped[int] = mapped_column(Integer, default=0)
user: Mapped["User"] = relationship(back_populates="user_tags")
tag: Mapped["Tag"] = relationship(back_populates="user_tags")
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(primary_key=True)
user_tags: Mapped[list["UserTag"]] = relationship(back_populates="user")
# Proxy для доступа к объектам тегов
tags = association_proxy('user_tags', 'tag')
# Proxy для доступа к score через tag_id
tag_scores = association_proxy('user_tags', 'score')
class Tag(Base):
__tablename__ = 'tags'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(50))
user_tags: Mapped[list["UserTag"]] = relationship(back_populates="tag")
# Использование
user = session.get(User, 1)
# Доступ к тегам через proxy
for tag in user.tags:
print(tag.name)
# Добавление тега через proxy
new_tag = Tag(name='python')
user.tags.append(new_tag)
session.commit()from sqlalchemy import Computed
class Employee(Base):
__tablename__ = 'employees'
id: Mapped[int] = mapped_column(primary_key=True)
salary: Mapped[int] = mapped_column(Integer)
bonus: Mapped[int] = mapped_column(Integer, default=0)
# Вычисляемая колонка на уровне БД
total_compensation: Mapped[int] = mapped_column(
Integer,
Computed('salary + bonus')
)
# С выражением
annual_salary: Mapped[int] = mapped_column(
Integer,
Computed('salary * 12')
)
# Использование
employee = Employee(salary=5000, bonus=1000)
session.add(employee)
session.commit()
session.refresh(employee)
print(employee.total_compensation) # 6000
print(employee.annual_salary) # 60000
# Запросы используют вычисляемую колонку
stmt = select(Employee).where(Employee.total_compensation > 50000)from sqlalchemy import Computed, case
class Product(Base):
__tablename__ = 'products'
id: Mapped[int] = mapped_column(primary_key=True)
price: Mapped[int] = mapped_column(Integer)
discount_percent: Mapped[int] = mapped_column(Integer, default=0)
is_sale: Mapped[bool] = mapped_column(Boolean, default=False)
# Условное вычисление
final_price: Mapped[int] = mapped_column(
Integer,
Computed(
case(
(is_sale == True, price * (100 - discount_percent) / 100),
else_=price
)
)
)class Category(Base):
__tablename__ = 'categories'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
parent_id: Mapped[int | None] = mapped_column(ForeignKey('categories.id'))
# Родитель
parent: Mapped["Category | None"] = relationship(
back_populates="children",
remote_side=[id],
lazy="joined"
)
# Дети
children: Mapped[list["Category"]] = relationship(
back_populates="parent",
lazy="selectin"
)
@hybrid_property
def full_path(self) -> str:
"""Полный путь категории (Python)."""
if self.parent:
return f"{self.parent.full_path} > {self.name}"
return self.name
# Загрузка дерева
from sqlalchemy.orm import selectinload
stmt = select(Category).options(
selectinload(Category.children).selectinload(Category.children)
)
categories = session.execute(stmt).scalars().all()class Student(Base):
__tablename__ = 'students'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
courses: Mapped[list["StudentCourse"]] = relationship(
back_populates="student"
)
@property
def enrolled_courses(self) -> list["Course"]:
"""Удобный доступ к курсам."""
return [sc.course for sc in self.courses]
class Course(Base):
__tablename__ = 'courses'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
students: Mapped[list["StudentCourse"]] = relationship(
back_populates="course"
)
class StudentCourse(Base):
"""Ассоциативная модель с дополнительными полями."""
__tablename__ = 'student_courses'
student_id: Mapped[int] = mapped_column(ForeignKey('students.id'), primary_key=True)
course_id: Mapped[int] = mapped_column(ForeignKey('courses.id'), primary_key=True)
enrolled_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
grade: Mapped[str | None] = mapped_column(String(2))
attendance: Mapped[int] = mapped_column(Integer, default=0)
student: Mapped["Student"] = relationship(back_populates="courses")
course: Mapped["Course"] = relationship(back_populates="students")
# Использование
student = Student(name='Alice')
course = Course(title='Math')
enrollment = StudentCourse(
student=student,
course=course,
grade='A',
attendance=95
)
session.add(enrollment)
session.commit()
# Доступ
print(student.enrolled_courses) # [Course(title='Math')]
print(enrollment.grade) # 'A'class Employee(Base):
__tablename__ = 'employees'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
type: Mapped[str] = mapped_column(String(50))
__mapper_args__ = {
'polymorphic_on': type,
'polymorphic_identity': 'employee',
}
class Manager(Employee):
__mapper_args__ = {
'polymorphic_identity': 'manager',
}
department: Mapped[str] = mapped_column(String(100))
class Engineer(Employee):
__mapper_args__ = {
'polymorphic_identity': 'engineer',
}
programming_language: Mapped[str] = mapped_column(String(100))
# Использование
manager = Manager(name='Bob', department='Engineering')
engineer = Engineer(name='Alice', programming_language='Python')
session.add_all([manager, engineer])
session.commit()
# Запрос возвращает правильные подклассы
employees = session.query(Employee).all()
print(type(employees[0])) # <class 'Manager'>
print(type(employees[1])) # <class 'Engineer'># Используйте hybrid_property для атрибутов которые нужны и в Python и в SQL
@hybrid_property
def full_name(self):
return f"{self.first_name} {self.last_name}"
@full_name.expression
def full_name(cls):
return func.concat(cls.first_name, ' ', cls.last_name)
# Используйте association_proxy для удобного доступа к связям
tags = association_proxy('user_tags', 'tag')
# Используйте Computed для вычислений на уровне БД
total = mapped_column(Integer, Computed('price * quantity'))
# Используйте remote_side для self-referential связей
parent = relationship("Category", back_populates="children", remote_side=[id])# Не используйте hybrid_property для сложных вычислений в SQL
# Если выражение сложное, лучше создать явный метод для запросов
# Не забывайте .expression для hybrid_property
# Без него нельзя использовать в запросах
# Не используйте association_proxy для связей с важными дополнительными полями
# Лучше явный доступ через ассоциативную модель
# Не вычисляйте в Python то что можно вычислить в БД
# Computed колонки эффективнееДалее: Polymorphism