Bases de Données Python — Cours Complet

1. SQLAlchemy 2.0 — ORM Moderne

Configuration

pip install sqlalchemy asyncpg alembic

Declarative Base — Modèles

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, Integer, Float, ForeignKey, DateTime
from typing import Optional
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    # Relationships
    orders: Mapped[list["Order"]] = relationship(back_populates="user")

    def __repr__(self) -> str:
        return f"User(id={self.id}, name={self.name})"

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    total: Mapped[float] = mapped_column(Float)
    status: Mapped[str] = mapped_column(String(20), default="pending")
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    user: Mapped["User"] = relationship(back_populates="orders")
    items: Mapped[list["OrderItem"]] = relationship(back_populates="order")

class OrderItem(Base):
    __tablename__ = "order_items"

    id: Mapped[int] = mapped_column(primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"))
    product_name: Mapped[str] = mapped_column(String(200))
    quantity: Mapped[int] = mapped_column(Integer)
    price: Mapped[float] = mapped_column(Float)

    order: Mapped["Order"] = relationship(back_populates="items")

Session — CRUD

from sqlalchemy import create_engine, select, delete
from sqlalchemy.orm import Session

engine = create_engine("postgresql://user:pass@localhost/db", echo=True)

# CREATE
with Session(engine) as session:
    user = User(name="Alice", email="alice@example.com")
    session.add(user)
    session.commit()
    print(f"Created user with id {user.id}")

# READ
with Session(engine) as session:
    # Tous les utilisateurs
    users = session.execute(select(User)).scalars().all()

    # Par filtre
    user = session.execute(
        select(User).where(User.email == "alice@example.com")
    ).scalar_one_or_none()

    # Par clé primaire
    user = session.get(User, 1)

# UPDATE
with Session(engine) as session:
    user = session.get(User, 1)
    if user:
        user.name = "Alice Updated"
        session.commit()

# DELETE
with Session(engine) as session:
    user = session.get(User, 1)
    if user:
        session.delete(user)
        session.commit()

Async Session

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

async_engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    echo=True,
    pool_size=5,
    max_overflow=10,
)

AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

async def create_user(name: str, email: str) -> User:
    async with AsyncSessionLocal() as session:
        user = User(name=name, email=email)
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

async def get_users() -> list[User]:
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User))
        return list(result.scalars().all())

Query Optimization — N+1 Problem

# N+1 Problem — Mauvais !
async def get_orders_naive() -> list:
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(Order))
        orders = result.scalars().all()
        # Pour chaque commande, une requête supplémentaire !
        for order in orders:
            print(order.user.name)  # N requêtes !

# Solution 1: eager loading (joined)
from sqlalchemy.orm import joinedload

async def get_orders_eager():
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            select(Order).options(joinedload(Order.user))
        )
        orders = result.unique().scalars().all()
        for order in orders:
            print(order.user.name)  # Pas de requête supplémentaire

# Solution 2: selectinload (pour relations multiples)
from sqlalchemy.orm import selectinload

async def get_orders_with_items():
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            select(Order).options(
                joinedload(Order.user),
                selectinload(Order.items),
            )
        )
        orders = result.unique().scalars().all()
        for order in orders:
            print(f"{order.user.name}: {len(order.items)} items")

Transactions

from sqlalchemy.exc import SQLAlchemyError

async def transfer_funds(from_id: int, to_id: int, amount: float) -> bool:
    async with AsyncSessionLocal() as session:
        try:
            async with session.begin():  # transaction
                sender = await session.get(User, from_id)
                receiver = await session.get(User, to_id)
                sender.balance -= amount
                receiver.balance += amount
            return True
        except SQLAlchemyError:
            return False

2. Alembic — Migrations

Configuration

alembic init alembic

# alembic/env.py
from models import Base
target_metadata = Base.metadata

Commandes

# Auto-générer une migration
alembic revision --autogenerate -m "add user table"

# Appliquer les migrations
alembic upgrade head

# Revenir en arrière
alembic downgrade -1

# Voir l'historique
alembic history

Migration Manuelle

"""add user table

Revision ID: abc123
Revises:
Create Date: 2024-01-01
"""
from alembic import op
import sqlalchemy as sa

def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("name", sa.String(100), nullable=False),
        sa.Column("email", sa.String(255), unique=True),
    )

def downgrade() -> None:
    op.drop_table("users")

3. asyncpg — PostgreSQL Driver

import asyncpg

class AsyncpgManager:
    """Gestionnaire de connexion PostgreSQL bas niveau."""

    def __init__(self, dsn: str):
        self.dsn = dsn
        self.pool: asyncpg.Pool | None = None

    async def connect(self):
        self.pool = await asyncpg.create_pool(
            self.dsn,
            min_size=5,
            max_size=20,
            command_timeout=60,
        )

    async def close(self):
        if self.pool:
            await self.pool.close()

    async def fetch(self, query: str, *args) -> list[asyncpg.Record]:
        async with self.pool.acquire() as conn:
            return await conn.fetch(query, *args)

    async def execute(self, query: str, *args) -> str:
        async with self.pool.acquire() as conn:
            return await conn.execute(query, *args)

# Usage
async def example():
    db = AsyncpgManager("postgresql://user:pass@localhost/db")
    await db.connect()
    try:
        rows = await db.fetch("SELECT * FROM users WHERE name = $1", "Alice")
        for row in rows:
            print(row["name"], row["email"])
    finally:
        await db.close()

4. Redis — Cache et Files d'Attente

redis-py

import redis.asyncio as redis
from typing import Optional
import json

class RedisCache:
    """Cache Redis asynchrone."""

    def __init__(self, url: str = "redis://localhost"):
        self.redis = None
        self.url = url

    async def connect(self):
        self.redis = await redis.from_url(
            self.url, decode_responses=True
        )

    async def get(self, key: str) -> Optional[str]:
        return await self.redis.get(key)

    async def set(self, key: str, value: str, ttl: int = 300):
        await self.redis.set(key, value, ex=ttl)

    async def delete(self, key: str):
        await self.redis.delete(key)

    async def cache_result(self, key: str, func, ttl: int = 300):
        """Cache-aside pattern."""
        cached = await self.get(key)
        if cached:
            return json.loads(cached)

        result = await func()
        await self.set(key, json.dumps(result), ttl)
        return result

# Redis queue
class TaskQueue:
    def __init__(self, redis_url: str):
        self.redis = None
        self.url = redis_url

    async def connect(self):
        self.redis = await redis.from_url(self.url)

    async def push(self, queue: str, task: dict):
        await self.redis.lpush(queue, json.dumps(task))

    async def pop(self, queue: str, timeout: int = 5) -> Optional[dict]:
        result = await self.redis.brpop(queue, timeout=timeout)
        if result:
            _, data = result
            return json.loads(data)
        return None

5. MongoDB avec Beanie ODM

from beanie import Document, Indexed, init_beanie
from motor.motor_asyncio import AsyncIOMotorClient
from typing import Optional
from datetime import datetime

# Document Beanie (ODM)
class Product(Document):
    name: str
    price: float
    category: str
    tags: list[str] = []
    in_stock: bool = True
    created_at: datetime = datetime.utcnow()

    class Settings:
        name = "products"
        indexes = [
            "name",
            [("category", 1), ("price", -1)],  # index composé
        ]

# Initialisation
async def init_mongo():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    await init_beanie(database=client.mydb, document_models=[Product])

# CRUD Beanie
async def crud_example():
    # Create
    product = Product(name="Laptop", price=999.99, category="electronics")
    await product.insert()

    # Find
    laptop = await Product.find_one(Product.name == "Laptop")
    cheap = await Product.find(Product.price < 500).to_list()

    # Update
    await product.set({Product.price: 899.99})

    # Aggregation
    pipeline = [
        {"$group": {"_id": "$category", "avg_price": {"$avg": "$price"}}}
    ]
    results = await Product.aggregate(pipeline).to_list()

6. NoSQL vs SQL — Quand Choisir

7. Bonnes Pratiques

# 1. Connection pooling
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,  # vérifie la connexion avant usage
)

# 2. Sessions comme dépendances FastAPI
async def get_db() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        yield session

# 3. N+1 detection
import logging
logging.getLogger("sqlalchemy.engine").setLevel(logging.DEBUG)

# 4. Batch operations
async def bulk_insert(items: list[dict]):
    async with AsyncSessionLocal() as session:
        session.add_all([User(**item) for item in items])
        await session.commit()

# 5. Raw SQL quand nécessaire
result = await session.execute(
    text("SELECT * FROM users WHERE name LIKE :name"),
    {"name": "%Alice%"},
)