Какой уровень запросов делал в SQLAlchemy?

SQLAlchemy: уровни запросов

В SQLAlchemy есть три основных уровня работы с БД: Core, ORM и выполнение сырого SQL. Я использую все три в зависимости от задачи.

1. SQLAlchemy ORM (Object-Relational Mapping)

Самый высокоуровневый подход. Работаю с объектами Python вместо SQL.

Определение модели:

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    posts = relationship('Post', back_populates='author')

class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.utcnow)
    
    author = relationship('User', back_populates='posts')

Использование ORM:

from sqlalchemy.orm import Session

def create_user(db: Session, username: str, email: str) -> User:
    user = User(username=username, email=email)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

def get_user_by_id(db: Session, user_id: int) -> User:
    return db.query(User).filter(User.id == user_id).first()

def get_all_posts_by_user(db: Session, user_id: int) -> list:
    return db.query(Post).filter(Post.user_id == user_id).all()

# Eager loading для избежания N+1 problem
from sqlalchemy.orm import joinedload

def get_user_with_posts(db: Session, user_id: int) -> User:
    return db.query(User).options(
        joinedload(User.posts)
    ).filter(User.id == user_id).first()

Плюсы ORM:

• Безопасность от SQL injection
• Абстракция от БД (легче менять БД)
• Автоматическое управление отношениями
• Типизация в Python

Минусы ORM:

• Медленнее сырого SQL в сложных запросах
• N+1 проблемы, если неправильно использовать
• Сложнее оптимизировать

2. SQLAlchemy Core (expression language)

Промежуточный уровень. Работаю с выражениями, а не объектами.

from sqlalchemy import select, insert, update, delete, and_, or_
from sqlalchemy.orm import Session

# SELECT
stmt = select(User).where(User.email.like('%@gmail.com'))
users = db.execute(stmt).scalars().all()

# INSERT
stmt = insert(User).values(
    username='john',
    email='john@example.com'
)
result = db.execute(stmt)
db.commit()

# UPDATE
stmt = update(User).where(
    User.id == 5
).values(
    email='newemail@example.com'
)
db.execute(stmt)
db.commit()

# DELETE
stmt = delete(User).where(User.id == 5)
db.execute(stmt)
db.commit()

# Сложные условия
stmt = select(User).where(
    and_(
        User.email.like('%@gmail.com'),
        User.created_at > some_date
    )
).order_by(User.created_at.desc())

JOIN в Core:

from sqlalchemy import join

stmt = select(User, Post).join(
    Post, User.id == Post.user_id
).where(User.id == 1)

results = db.execute(stmt).all()

3. Сырой SQL (Raw SQL)

Для сложных запросов или специфичных оптимизаций.

from sqlalchemy import text

# Простой запрос
stmt = text('SELECT * FROM users WHERE email = :email')
result = db.execute(stmt, {'email': 'john@example.com'}).first()

# Сложный запрос с подзапросами
query = text('''
    SELECT u.id, u.username, COUNT(p.id) as post_count
    FROM users u
    LEFT JOIN posts p ON u.id = p.user_id
    WHERE u.created_at > :date
    GROUP BY u.id
    ORDER BY post_count DESC
    LIMIT 10
''')

results = db.execute(query, {'date': '2024-01-01'}).fetchall()

Мой практический подход

ORM использую для:

• CRUD операции (create, read, update, delete)
• Простые фильтры и поиск
• Работа с отношениями
• Быстрое прототипирование

Core использую для:

• Bulk операции (много данных за раз)
• Сложные условия
• JOINs со множеством таблиц
• Когда нужна большая производительность

Raw SQL использую для:

• Window functions
• CTEs (Common Table Expressions)
• Специфичные оптимизации БД
• Миграции (всегда сырой SQL)

Реальный пример из production

from sqlalchemy import and_, or_, func
from sqlalchemy.orm import Session

def search_products(db: Session, query: str, category: str, min_price: float):
    # Комбинирую ORM и Core
    stmt = select(Product).where(
        and_(
            or_(
                Product.name.ilike(f'%{query}%'),
                Product.description.ilike(f'%{query}%')
            ),
            Product.category == category,
            Product.price >= min_price
        )
    ).order_by(
        Product.rating.desc()
    ).limit(50)
    
    return db.execute(stmt).scalars().all()

# Для analytics — raw SQL
def get_daily_sales(db: Session, start_date, end_date):
    query = text('''
        SELECT DATE(created_at) as date, SUM(amount) as total
        FROM orders
        WHERE created_at BETWEEN :start AND :end
        GROUP BY DATE(created_at)
        ORDER BY date ASC
    ''')
    
    return db.execute(
        query,
        {'start': start_date, 'end': end_date}
    ).fetchall()

Асинхронность (async SQLAlchemy)

В современных проектах часто использую async:

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

engine = create_async_engine('postgresql+asyncpg://user:pass@localhost/db')

async def get_user_async(session: AsyncSession, user_id: int):
    stmt = select(User).where(User.id == user_id)
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

Выбор уровня зависит от баланса между удобством (ORM) и производительностью (Raw SQL). Обычно начинаю с ORM, затем оптимизирую при необходимости.