SQLAlchemy vs Django ORM: работа с БД

1from django.db import models
2
3class Category(models.Model):
4    name = models.CharField(max_length=100)
5    slug = models.SlugField(unique=True)
6    created_at = models.DateTimeField(auto_now_add=True)
7
8    class Meta:
9        verbose_name_plural = "categories"
10
11    def __str__(self):
12        return self.name
13
14class Product(models.Model):
15    name = models.CharField(max_length=200)
16    description = models.TextField(blank=True)
17    price = models.DecimalField(max_digits=10, decimal_places=2)
18    category = models.ForeignKey(Category, on_delete=models.CASCADE, related_name='products')
19    is_active = models.BooleanField(default=True)
20    created_at = models.DateTimeField(auto_now_add=True)
21
22    def __str__(self):
23        return self.name

1# Получение всех записей
2products = Product.objects.all()
3
4# Фильтрация
5active_products = Product.objects.filter(is_active=True)
6
7# Сложные запросы
8from django.db.models import Q, F
9products = Product.objects.filter(
10    Q(category__name='Electronics') & Q(price__lt=100)
11)
12
13# Агрегация
14from django.db.models import Avg, Count, Sum
15stats = Product.objects.aggregate(
16    avg_price=Avg('price'),
17    total_count=Count('id'),
18    total_value=Sum(F('price') * F('quantity'))
19)
20
21# Асинхронные запросы (Django 4.1+)
22async def get_products():
23    products = await Product.objects.filter(is_active=True).all()
24    return products

1# Создание миграций
2python manage.py makemigrations
3
4# Применение миграций
5python manage.py migrate
6
7# Откат миграций
8python manage.py migrate myapp 0001

1from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, DateTime
2from sqlalchemy.orm import declarative_base, relationship, Session
3from datetime import datetime
4
5Base = declarative_base()
6
7class Category(Base):
8    __tablename__ = 'categories'
9
10    id = Column(Integer, primary_key=True)
11    name = Column(String(100), nullable=False)
12    slug = Column(String(100), unique=True, nullable=False)
13    created_at = Column(DateTime, default=datetime.utcnow)
14
15    products = relationship("Product", back_populates="category")
16
17class Product(Base):
18    __tablename__ = 'products'
19
20    id = Column(Integer, primary_key=True)
21    name = Column(String(200), nullable=False)
22    description = Column(String, default='')
23    price = Column(Integer, nullable=False)  # Храним в копейках
24    category_id = Column(Integer, ForeignKey('categories.id'))
25
26    category = relationship("Category", back_populates="products")

1from sqlalchemy import select, func, and_, or_
2from sqlalchemy.orm import selectinload
3
4# Простой запрос
5stmt = select(Product).where(Product.is_active == True)
6products = session.execute(stmt).scalars().all()
7
8# Сложные запросы
9stmt = (
10    select(Product)
11    .options(selectinload(Product.category))  # Eager loading
12    .where(
13        and_(
14            Product.category.has(name='Electronics'),
15            Product.price < 10000  # 100 рублей в копейках
16        )
17    )
18    .order_by(Product.price.desc())
19    .limit(10)
20)
21
22# Агрегация
23stmt = (
24    select(
25        Product.category_id,
26        func.count(Product.id).label('count'),
27        func.avg(Product.price).label('avg_price')
28    )
29    .group_by(Product.category_id)
30    .having(func.count(Product.id) > 5)
31)

1from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
2from sqlalchemy.orm import sessionmaker
3
4engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
5async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
6
7async def get_products():
8    async with async_session() as session:
9        result = await session.execute(
10            select(Product).where(Product.is_active == True)
11        )
12        return result.scalars().all()