Single/multi/joined table inheritance, polymorphic loading
Полиморфизм в SQLAlchemy позволяет работать с иерархиями классов где разные подклассы сохраняются в базу данных. Вы изучите single, joined и multi-table inheritance паттерны.
Все подклассы хранятся в одной таблице с discriminator колонкой.
from sqlalchemy import String, Integer, Boolean
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
class Base(DeclarativeBase):
pass
class Employee(Base):
"""Базовый класс для всех сотрудников."""
__tablename__ = 'employees'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
# Discriminator колонка
type: Mapped[str] = mapped_column(String(50))
# Общие поля для всех подклассов
hire_date: Mapped[datetime] = mapped_column(DateTime)
salary: Mapped[int] = mapped_column(Integer)
__mapper_args__ = {
'polymorphic_on': type, # Колонка для определения типа
'polymorphic_identity': 'employee', # Значение для базового класса
}
def __repr__(self):
return f"<Employee(name={self.name}, type={self.type})>"
class Manager(Employee):
"""Менеджеры."""
__tablename__ = 'employees' # Та же таблица
# Специфичные поля менеджера
department: Mapped[str] = mapped_column(String(100))
bonus: Mapped[int] = mapped_column(Integer, default=0)
__mapper_args__ = {
'polymorphic_identity': 'manager', # Значение для этого подкласса
}
def __repr__(self):
return f"<Manager(name={self.name}, department={self.department})>"
class Engineer(Employee):
"""Инженеры."""
__tablename__ = 'employees' # Та же таблица
# Специфичные поля инженера
programming_language: Mapped[str] = mapped_column(String(100))
level: Mapped[int] = mapped_column(Integer, default=1)
__mapper_args__ = {
'polymorphic_identity': 'engineer',
}
def __repr__(self):
return f"<Engineer(name={self.name}, language={self.programming_language})>"# Создание объектов разных типов
manager = Manager(
name='Alice',
hire_date=datetime(2020, 1, 1),
salary=100000,
department='Engineering'
)
engineer = Engineer(
name='Bob',
hire_date=datetime(2021, 6, 1),
salary=80000,
programming_language='Python'
)
session.add_all([manager, engineer])
session.commit()
# Запрос возвращает правильные подклассы
employees = session.query(Employee).all()
print(type(employees[0])) # <class '__main__.Manager'>
print(type(employees[1])) # <class '__main__.Engineer'>
# Фильтрация по типу
managers = session.query(Employee).filter(Employee.type == 'manager').all()
# Или через isinstance
engineers = [e for e in employees if isinstance(e, Engineer)]| Преимущества | Недостатки |
|---|---|
| Быстрые запросы (нет JOIN) | Таблица может стать широкой |
| Простая схема БД | Много NULL для специфичных полей |
| Легко добавить новый подкласс | Сложно добавить обязательное поле для подкласса |
Каждый подкласс имеет свою таблицу с FK на родительскую таблицу.
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))
hire_date: Mapped[datetime] = mapped_column(DateTime)
salary: Mapped[int] = mapped_column(Integer)
__mapper_args__ = {
'polymorphic_on': type,
'polymorphic_identity': 'employee',
}
class Manager(Employee):
"""Таблица менеджеров."""
__tablename__ = 'managers'
# FK на родительскую таблицу
id: Mapped[int] = mapped_column(
ForeignKey('employees.id'),
primary_key=True
)
# Специфичные поля
department: Mapped[str] = mapped_column(String(100))
bonus: Mapped[int] = mapped_column(Integer, default=0)
__mapper_args__ = {
'polymorphic_identity': 'manager',
}
class Engineer(Employee):
"""Таблица инженеров."""
__tablename__ = 'engineers'
id: Mapped[int] = mapped_column(
ForeignKey('employees.id'),
primary_key=True
)
programming_language: Mapped[str] = mapped_column(String(100))
level: Mapped[int] = mapped_column(Integer, default=1)
__mapper_args__ = {
'polymorphic_identity': 'engineer',
}"""Create employee tables with joined inheritance
Revision ID: joined_inh_001
Revises: prev_001
"""
from alembic import op
import sqlalchemy as sa
revision = 'joined_inh_001'
down_revision = 'prev_001'
def upgrade():
# Базовая таблица
op.create_table(
'employees',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(100), nullable=False),
sa.Column('type', sa.String(50), nullable=False),
sa.Column('hire_date', sa.DateTime(), nullable=False),
sa.Column('salary', sa.Integer(), nullable=False),
sa.PrimaryKeyConstraint('id'),
)
# Таблица менеджеров
op.create_table(
'managers',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('department', sa.String(100), nullable=False),
sa.Column('bonus', sa.Integer(), default=0),
sa.ForeignKeyConstraint(['id'], ['employees.id']),
sa.PrimaryKeyConstraint('id'),
)
# Таблица инженеров
op.create_table(
'engineers',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('programming_language', sa.String(100), nullable=False),
sa.Column('level', sa.Integer(), default=1),
sa.ForeignKeyConstraint(['id'], ['employees.id']),
sa.PrimaryKeyConstraint('id'),
)
def downgrade():
op.drop_table('engineers')
op.drop_table('managers')
op.drop_table('employees')| Преимущества | Недостатки |
|---|---|
| Нормализованная схема | Запросы требуют JOIN |
| Нет NULL для специфичных полей | Сложнее запросы |
| Легко добавить обязательные поля | Медленнее при большом количестве подклассов |
Каждый подкласс имеет свою таблицу со всеми полями (включая унаследованные).
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))
hire_date: Mapped[datetime] = mapped_column(DateTime)
salary: Mapped[int] = mapped_column(Integer)
__mapper_args__ = {
'polymorphic_on': type,
'concrete': True, # Concrete inheritance
}
class Manager(Employee):
__tablename__ = 'managers'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
type: Mapped[str] = mapped_column(String(50))
hire_date: Mapped[datetime] = mapped_column(DateTime)
salary: Mapped[int] = mapped_column(Integer)
# Специфичные поля
department: Mapped[str] = mapped_column(String(100))
bonus: Mapped[int] = mapped_column(Integer, default=0)
__mapper_args__ = {
'polymorphic_identity': 'manager',
'concrete': True,
}
class Engineer(Employee):
__tablename__ = 'engineers'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
type: Mapped[str] = mapped_column(String(50))
hire_date: Mapped[datetime] = mapped_column(DateTime)
salary: Mapped[int] = mapped_column(Integer)
programming_language: Mapped[str] = mapped_column(String(100))
level: Mapped[int] = mapped_column(Integer, default=1)
__mapper_args__ = {
'polymorphic_identity': 'engineer',
'concrete': True,
}| Преимущества | Недостатки |
|---|---|
| Полная независимость таблиц | Дублирование схемы |
| Нет JOIN для запросов подкласса | Сложно запросить все подклассы |
| Можно оптимизировать каждую таблицу | Изменение общего поля требует миграции всех таблиц |
from sqlalchemy.orm import polymorphic_load
# По умолчанию: загружает все подклассы правильно
employees = session.query(Employee).all()
# Явное указание стратегии
stmt = select(Employee).options(
polymorphic_load(Employee) # Загрузить все подклассы
)
employees = session.execute(stmt).scalars().all()from sqlalchemy.orm import selectinload
# Для joined inheritance может потребоваться selectinload
stmt = select(Employee).options(
selectinload(Manager.department), # Подгрузка специфичных полей
)
employees = session.execute(stmt).scalars().all()# Через discriminator
managers = session.query(Employee).filter(
Employee.type == 'manager'
).all()
# Через isinstance
from sqlalchemy.orm import aliased
engineers = session.query(Engineer).all()
# Через with_polymorphic
from sqlalchemy.orm import with_polymorphic
emp_poly = with_polymorphic(Employee, [Manager, Engineer])
stmt = select(emp_poly).where(
emp_poly.Engineer.programming_language == 'Python'
)
engineers = session.execute(stmt).scalars().all()from sqlalchemy import func
# Средняя зарплата по типам
stmt = select(
Employee.type,
func.avg(Employee.salary).label('avg_salary')
).group_by(Employee.type)
result = session.execute(stmt).all()
# [('manager', 95000), ('engineer', 75000)]
# Количество сотрудников по департаментам (только для менеджеров)
stmt = select(
Manager.department,
func.count(Manager.id).label('count')
).group_by(Manager.department)
dept_counts = session.execute(stmt).all()# Используйте Single Table для простых иерархий сfew подклассами
# Используйте Joined Table для сложных иерархий с many специфичными полями
# Добавляйте индекс на discriminator колонку
type: Mapped[str] = mapped_column(String(50), index=True)
# Используйте with_polymorphic для эффективных запросов
emp_poly = with_polymorphic(Employee, [Manager, Engineer])
# Документируйте иерархию в моделях
class Employee(Base):
"""Базовый класс для иерархии сотрудников.
Подклассы: Manager, Engineer, Designer
"""# Не создавайте слишком глубокую иерархию (>3 уровней)
# Это усложняет запросы и поддержку
# Не используйте Concrete inheritance если нужно часто
# запрашивать все подклассы вместе
# Не забывайте про polymorphic_identity
# Без него SQLAlchemy не сможет определить тип
# Не изменяйте discriminator колонку вручную
# type = 'manager' # ПЛОХО: используйте правильный подклассВ следующей теме вы изучите Multi-DB и Sharding — горизонтальное масштабирование, routing запросов, sharding стратегии и управление несколькими базами данных.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.