
# Fichier: python_cheats/cheatsheets/sqlalchemy_complete.txt
# SQLAlchemy - Cheatsheet Complète et Avancée



[OK] INSTALLATION ET IMPORTS


[OK] INSTALLATION
    pip install sqlalchemy
    pip install sqlalchemy[asyncio]  # Pour le support async
    pip install alembic              # Pour les migrations

[OK] IMPORTS ESSENTIELS
    from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean
    from sqlalchemy import ForeignKey, Table, Index, CheckConstraint, UniqueConstraint
    from sqlalchemy import DateTime, Date, Time, Text, JSON, ARRAY, Enum
    from sqlalchemy import func, and_, or_, not_, case, cast, literal
    from sqlalchemy.orm import sessionmaker, Session, relationship, backref
    from sqlalchemy.orm import declarative_base, selectinload, joinedload, subqueryload
    from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
    from sqlalchemy.ext.associationproxy import association_proxy
    from sqlalchemy.pool import NullPool, QueuePool, StaticPool
    from sqlalchemy.sql import select, insert, update, delete
    from datetime import datetime, date


[OK] CONNEXION À LA BASE DE DONNÉES


[OK] DIFFÉRENTS TYPES DE CONNEXION
    # SQLite (fichier)
    engine = create_engine('sqlite:///database.db', echo=True)
    
    # SQLite (mémoire)
    engine = create_engine('sqlite:///:memory:', echo=True)
    
    # PostgreSQL
    engine = create_engine('postgresql://user:password@localhost:5432/dbname')
    engine = create_engine('postgresql+psycopg2://user:pass@localhost/dbname')
    
    # MySQL
    engine = create_engine('mysql+pymysql://user:password@localhost/dbname')
    engine = create_engine('mysql+mysqlconnector://user:pass@localhost/dbname')
    
    # Microsoft SQL Server
    engine = create_engine('mssql+pyodbc://user:pass@server/dbname?driver=ODBC+Driver+17+for+SQL+Server')
    
    # Oracle
    engine = create_engine('oracle+cx_oracle://user:pass@localhost:1521/dbname')

[OK] OPTIONS DE CONNEXION AVANCÉES
    engine = create_engine(
        'postgresql://user:pass@localhost/dbname',
        echo=True,                    # Afficher les requêtes SQL
        echo_pool=True,               # Logs du pool de connexions
        pool_size=5,                  # Nombre de connexions dans le pool
        max_overflow=10,              # Connexions supplémentaires autorisées
        pool_timeout=30,              # Timeout pour obtenir une connexion
        pool_recycle=3600,            # Recycler les connexions après 1h
        pool_pre_ping=True,           # Vérifier la connexion avant utilisation
        connect_args={                # Arguments spécifiques au driver
            'connect_timeout': 10,
            'application_name': 'MyApp'
        }
    )

[OK] POOL DE CONNEXIONS
    # Pool par défaut (QueuePool)
    engine = create_engine('postgresql://...', poolclass=QueuePool)
    
    # Pas de pool (nouvelle connexion à chaque fois)
    engine = create_engine('postgresql://...', poolclass=NullPool)
    
    # Pool statique (threads)
    engine = create_engine('sqlite://...', poolclass=StaticPool)


[OK] CRÉATION DE SESSION


[OK] SESSION CLASSIQUE
    from sqlalchemy.orm import sessionmaker
    
    Session = sessionmaker(bind=engine)
    session = Session()
    
    # Ou directement
    session = Session(engine)

[OK] SESSION AVEC CONTEXT MANAGER
    from sqlalchemy.orm import Session
    
    with Session(engine) as session:
        # Vos opérations
        session.commit()

[OK] SESSION SCOPED (Thread-safe)
    from sqlalchemy.orm import scoped_session, sessionmaker
    
    session_factory = sessionmaker(bind=engine)
    Session = scoped_session(session_factory)
    
    # Utilisation
    session = Session()
    # ... opérations
    Session.remove()  # Nettoyer après utilisation

[OK] SESSION AVEC AUTO-COMMIT DÉSACTIVÉ
    Session = sessionmaker(bind=engine, autocommit=False, autoflush=False)


[OK] DÉFINITION DES MODÈLES (ORM)


[OK] MODÈLE BASIQUE
    from sqlalchemy.ext.declarative import declarative_base
    
    Base = declarative_base()
    
    class User(Base):
        __tablename__ = 'users'
        
        id = Column(Integer, primary_key=True, autoincrement=True)
        username = Column(String(50), nullable=False, unique=True)
        email = Column(String(100), nullable=False, unique=True)
        password_hash = Column(String(255), nullable=False)
        age = Column(Integer, CheckConstraint('age >= 0'))
        is_active = Column(Boolean, default=True)
        created_at = Column(DateTime, default=datetime.utcnow)
        updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
        
        def __repr__(self):
            return f"<User(id={self.id}, username='{self.username}')>"

[OK] TYPES DE COLONNES AVANCÉS
    class Product(Base):
        __tablename__ = 'products'
        
        id = Column(Integer, primary_key=True)
        name = Column(String(100), nullable=False)
        price = Column(Float(precision=2), nullable=False)
        description = Column(Text)
        stock = Column(Integer, default=0)
        
        # Types spéciaux
        tags = Column(ARRAY(String))              # PostgreSQL
        metadata_json = Column(JSON)              # JSON
        status = Column(Enum('draft', 'published', 'archived', name='product_status'))
        
        # Dates et temps
        available_from = Column(Date)
        available_until = Column(Date)
        created_time = Column(Time)
        last_modified = Column(DateTime, default=datetime.utcnow)

[OK] CONTRAINTES
    class Account(Base):
        __tablename__ = 'accounts'
        
        id = Column(Integer, primary_key=True)
        account_number = Column(String(20), nullable=False)
        balance = Column(Float, default=0.0)
        
        # Contrainte CHECK
        __table_args__ = (
            CheckConstraint('balance >= 0', name='positive_balance'),
            CheckConstraint('length(account_number) = 20', name='valid_account_number'),
            UniqueConstraint('account_number', name='unique_account'),
            Index('idx_account_balance', 'balance'),
            Index('idx_account_composite', 'account_number', 'balance'),
        )

[OK] VALEURS PAR DÉFAUT AVANCÉES
    class Article(Base):
        __tablename__ = 'articles'
        
        id = Column(Integer, primary_key=True)
        title = Column(String(200), nullable=False)
        
        # Valeur par défaut simple
        views = Column(Integer, default=0)
        
        # Valeur par défaut avec fonction
        created_at = Column(DateTime, default=datetime.utcnow)
        
        # Valeur par défaut côté serveur (SQL)
        updated_at = Column(DateTime, server_default=func.now())
        
        # UUID par défaut
        import uuid
        uuid = Column(String(36), default=lambda: str(uuid.uuid4()))


[OK] RELATIONS ENTRE TABLES


[OK] ONE-TO-MANY (Un à Plusieurs)
    class Author(Base):
        __tablename__ = 'authors'
        
        id = Column(Integer, primary_key=True)
        name = Column(String(100))
        
        # Relation
        books = relationship('Book', back_populates='author', cascade='all, delete-orphan')
    
    class Book(Base):
        __tablename__ = 'books'
        
        id = Column(Integer, primary_key=True)
        title = Column(String(200))
        author_id = Column(Integer, ForeignKey('authors.id'))
        
        # Relation inverse
        author = relationship('Author', back_populates='books')

[OK] MANY-TO-MANY (Plusieurs à Plusieurs)
    # Table d'association
    student_course = Table('student_course', Base.metadata,
        Column('student_id', Integer, ForeignKey('students.id'), primary_key=True),
        Column('course_id', Integer, ForeignKey('courses.id'), primary_key=True),
        Column('enrollment_date', DateTime, default=datetime.utcnow)
    )
    
    class Student(Base):
        __tablename__ = 'students'
        
        id = Column(Integer, primary_key=True)
        name = Column(String(100))
        
        courses = relationship('Course', secondary=student_course, back_populates='students')
    
    class Course(Base):
        __tablename__ = 'courses'
        
        id = Column(Integer, primary_key=True)
        title = Column(String(200))
        
        students = relationship('Student', secondary=student_course, back_populates='courses')

[OK] ONE-TO-ONE (Un à Un)
    class Person(Base):
        __tablename__ = 'persons'
        
        id = Column(Integer, primary_key=True)
        name = Column(String(100))
        
        passport = relationship('Passport', uselist=False, back_populates='person')
    
    class Passport(Base):
        __tablename__ = 'passports'
        
        id = Column(Integer, primary_key=True)
        number = Column(String(20), unique=True)
        person_id = Column(Integer, ForeignKey('persons.id'), unique=True)
        
        person = relationship('Person', back_populates='passport')

[OK] SELF-REFERENTIAL (Auto-référence)
    class Employee(Base):
        __tablename__ = 'employees'
        
        id = Column(Integer, primary_key=True)
        name = Column(String(100))
        manager_id = Column(Integer, ForeignKey('employees.id'))
        
        # Manager (un seul)
        manager = relationship('Employee', remote_side=[id], backref='subordinates')

[OK] OPTIONS DE CASCADE
    # Cascade complète
    posts = relationship('Post', cascade='all, delete, delete-orphan')
    
    # Cascade partielle
    posts = relationship('Post', cascade='save-update, merge')
    
    # Options de cascade:
    # - all: tous les cascades sauf delete-orphan
    # - save-update: persist les nouveaux objets
    # - merge: merge les objets détachés
    # - delete: supprime les objets liés
    # - delete-orphan: supprime les objets orphelins
    # - expunge: retire de la session
    # - refresh-expire: rafraîchit les objets


[OK] INSERTION DE DONNÉES


[OK] INSERTION SIMPLE
    user = User(username='alice', email='alice@example.com', age=30)
    session.add(user)
    session.commit()

[OK] INSERTION MULTIPLE
    users = [
        User(username='bob', email='bob@example.com', age=25),
        User(username='charlie', email='charlie@example.com', age=35),
        User(username='diana', email='diana@example.com', age=28)
    ]
    session.add_all(users)
    session.commit()

[OK] INSERTION AVEC FLUSH (Sans commit)
    user = User(username='eve', email='eve@example.com')
    session.add(user)
    session.flush()  # Génère l'ID sans commit
    print(f"Nouvel ID: {user.id}")

[OK] INSERTION BULK (Performance)
    # Bulk insert (plus rapide, pas de suivi ORM)
    session.bulk_insert_mappings(User, [
        {'username': 'user1', 'email': 'user1@example.com'},
        {'username': 'user2', 'email': 'user2@example.com'},
        {'username': 'user3', 'email': 'user3@example.com'}
    ])
    session.commit()

[OK] INSERTION AVEC RETURNING
    from sqlalchemy.dialects.postgresql import insert
    
    stmt = insert(User).values(username='frank', email='frank@example.com').returning(User.id)
    result = session.execute(stmt)
    new_id = result.scalar()


[OK] REQUÊTES SELECT


[OK] REQUÊTES BASIQUES
    # Tous les résultats
    users = session.query(User).all()
    
    # Premier résultat
    user = session.query(User).first()
    
    # Un seul résultat (erreur si plusieurs ou aucun)
    user = session.query(User).one()
    
    # Un seul ou None
    user = session.query(User).one_or_none()
    
    # Par clé primaire
    user = session.query(User).get(1)
    user = session.get(User, 1)  # Nouveau style (SQLAlchemy 2.0)

[OK] FILTRAGE
    # Méthode 1: filter()
    users = session.query(User).filter(User.age > 25).all()
    
    # Méthode 2: filter_by() (uniquement pour égalité)
    users = session.query(User).filter_by(username='alice').all()
    
    # Multiples conditions
    users = session.query(User).filter(
        User.age > 18,
        User.is_active == True
    ).all()

[OK] OPÉRATEURS DE COMPARAISON
    # Égalité
    users = session.query(User).filter(User.age == 30).all()
    
    # Inégalité
    users = session.query(User).filter(User.age != 30).all()
    
    # Comparaisons
    users = session.query(User).filter(User.age > 25).all()
    users = session.query(User).filter(User.age >= 25).all()
    users = session.query(User).filter(User.age < 40).all()
    users = session.query(User).filter(User.age <= 40).all()
    
    # Between
    users = session.query(User).filter(User.age.between(25, 35)).all()
    
    # IN
    users = session.query(User).filter(User.age.in_([25, 30, 35])).all()
    
    # NOT IN
    users = session.query(User).filter(~User.age.in_([25, 30])).all()
    
    # IS NULL
    users = session.query(User).filter(User.email.is_(None)).all()
    
    # IS NOT NULL
    users = session.query(User).filter(User.email.isnot(None)).all()

[OK] FILTRAGE TEXTE (LIKE, ILIKE)
    # LIKE (sensible à la casse)
    users = session.query(User).filter(User.username.like('ali%')).all()
    
    # ILIKE (insensible à la casse, PostgreSQL)
    users = session.query(User).filter(User.username.ilike('%alice%')).all()
    
    # Commence par
    users = session.query(User).filter(User.username.startswith('al')).all()
    
    # Se termine par
    users = session.query(User).filter(User.username.endswith('ce')).all()
    
    # Contient
    users = session.query(User).filter(User.username.contains('lic')).all()

[OK] OPÉRATEURS LOGIQUES
    from sqlalchemy import and_, or_, not_
    
    # AND
    users = session.query(User).filter(
        and_(User.age > 25, User.is_active == True)
    ).all()
    
    # Équivalent (AND implicite)
    users = session.query(User).filter(User.age > 25, User.is_active == True).all()
    
    # OR
    users = session.query(User).filter(
        or_(User.age < 20, User.age > 60)
    ).all()
    
    # NOT
    users = session.query(User).filter(not_(User.is_active)).all()
    
    # Combinaison complexe
    users = session.query(User).filter(
        and_(
            User.age > 18,
            or_(User.username.like('a%'), User.email.like('%@gmail.com'))
        )
    ).all()

[OK] TRI (ORDER BY)
    # Ordre croissant
    users = session.query(User).order_by(User.age).all()
    
    # Ordre décroissant
    users = session.query(User).order_by(User.age.desc()).all()
    
    # Tri multiple
    users = session.query(User).order_by(User.age.desc(), User.username).all()
    
    # Tri avec NULL en premier/dernier
    users = session.query(User).order_by(User.email.nullsfirst()).all()
    users = session.query(User).order_by(User.email.nullslast()).all()

[OK] LIMITATION ET PAGINATION
    # Limiter le nombre de résultats
    users = session.query(User).limit(10).all()
    
    # Offset (sauter des résultats)
    users = session.query(User).offset(20).all()
    
    # Pagination
    page = 2
    per_page = 10
    users = session.query(User).offset((page - 1) * per_page).limit(per_page).all()
    
    # Slice (alternative pythonique)
    users = session.query(User)[10:20]  # Résultats 10 à 20

[OK] SÉLECTION DE COLONNES SPÉCIFIQUES
    # Sélectionner des colonnes spécifiques
    result = session.query(User.username, User.email).all()
    
    # Avec alias
    result = session.query(User.username.label('name')).all()
    
    # Tuple nommé
    from sqlalchemy.orm import Bundle
    
    bundle = Bundle('user_info', User.username, User.email)
    result = session.query(bundle).all()

[OK] DISTINCT
    # Résultats uniques
    ages = session.query(User.age).distinct().all()
    
    # Distinct sur plusieurs colonnes
    result = session.query(User.username, User.age).distinct().all()


[OK] UPDATE (Mise à jour)


[OK] UPDATE SIMPLE
    user = session.query(User).filter_by(username='alice').first()
    user.age = 31
    session.commit()

[OK] UPDATE EN MASSE
    # Mise à jour de plusieurs enregistrements
    session.query(User).filter(User.age < 18).update({'age': 18})
    session.commit()
    
    # Avec synchronize_session
    session.query(User).filter(User.age < 18).update(
        {'age': 18},
        synchronize_session='fetch'  # 'fetch', 'evaluate', ou False
    )
    session.commit()

[OK] UPDATE AVEC EXPRESSION
    # Incrémenter une valeur
    session.query(Product).filter(Product.id == 1).update(
        {'stock': Product.stock + 10}
    )
    session.commit()

[OK] UPDATE AVEC RETURNING (PostgreSQL)
    from sqlalchemy.dialects.postgresql import update
    
    stmt = (
        update(User)
        .where(User.id == 1)
        .values(age=User.age + 1)
        .returning(User.id, User.age)
    )
    result = session.execute(stmt)


[OK] DELETE (Suppression)


[OK] DELETE SIMPLE
    user = session.query(User).filter_by(username='alice').first()
    session.delete(user)
    session.commit()

[OK] DELETE EN MASSE
    session.query(User).filter(User.age < 18).delete()
    session.commit()
    
    # Avec synchronize_session
    session.query(User).filter(User.is_active == False).delete(
        synchronize_session='fetch'
    )
    session.commit()

[OK] TRUNCATE (Vider une table)
    # Attention: supprime toutes les données
    session.query(User).delete()
    session.commit()


[OK] JOINTURES


[OK] INNER JOIN
    # Méthode 1
    results = session.query(User, Post).join(Post).all()
    
    # Méthode 2 (explicite)
    results = session.query(User).join(Post, User.id == Post.user_id).all()
    
    # Méthode 3 (via relation)
    results = session.query(User).join(User.posts).all()

[OK] LEFT OUTER JOIN
    results = session.query(User).outerjoin(Post).all()

[OK] JOINTURES MULTIPLES
    results = (
        session.query(User)
        .join(Post)
        .join(Comment)
        .filter(Comment.content.like('%great%'))
        .all()
    )

[OK] SÉLECTIONNER DES COLONNES DE TABLES JOINTES
    results = session.query(
        User.username,
        Post.title,
        Post.created_at
    ).join(Post).all()

[OK] ALIAS POUR JOINTURES COMPLEXES
    from sqlalchemy.orm import aliased
    
    ParentUser = aliased(User)
    ChildUser = aliased(User)
    
    results = session.query(ParentUser, ChildUser).join(
        ChildUser, ParentUser.id == ChildUser.parent_id
    ).all()


[OK] AGRÉGATION ET FONCTIONS


[OK] FONCTIONS D'AGRÉGATION
    from sqlalchemy import func
    
    # COUNT
    count = session.query(func.count(User.id)).scalar()
    count = session.query(User).count()  # Alternative
    
    # SUM
    total = session.query(func.sum(Product.price)).scalar()
    
    # AVG
    average = session.query(func.avg(User.age)).scalar()
    
    # MIN et MAX
    min_age = session.query(func.min(User.age)).scalar()
    max_age = session.query(func.max(User.age)).scalar()

[OK] GROUP BY
    # Grouper par colonne
    results = session.query(
        User.age,
        func.count(User.id).label('count')
    ).group_by(User.age).all()
    
    # Grouper avec filtre HAVING
    results = session.query(
        User.age,
        func.count(User.id).label('count')
    ).group_by(User.age).having(func.count(User.id) > 5).all()

[OK] FONCTIONS DE CHAÎNES
    # Concaténation
    full_name = session.query(
        func.concat(User.first_name, ' ', User.last_name).label('full_name')
    ).all()
    
    # Majuscules/Minuscules
    upper = session.query(func.upper(User.username)).all()
    lower = session.query(func.lower(User.email)).all()
    
    # Longueur
    length = session.query(func.length(User.username)).all()

[OK] FONCTIONS DE DATE
    # Date actuelle
    now = func.now()
    current_date = func.current_date()
    
    # Extraction de parties de date
    year = session.query(func.extract('year', User.created_at)).all()
    month = session.query(func.extract('month', User.created_at)).all()
    
    # Différence de dates
    age_days = session.query(
        func.date_part('day', func.now() - User.created_at)
    ).all()

[OK] CASE WHEN
    from sqlalchemy import case
    
    result = session.query(
        User.username,
        case(
            (User.age < 18, 'Mineur'),
            (User.age < 65, 'Adulte'),
            else_='Senior'
        ).label('category')
    ).all()

[OK] CAST (Conversion de type)
    from sqlalchemy import cast
    
    result = session.query(
        cast(User.age, String)
    ).all()


[OK] SOUS-REQUÊTES


[OK] SOUS-REQUÊTE SCALAIRE
    # Sous-requête dans SELECT
    subq = session.query(func.avg(User.age)).scalar_subquery()
    
    results = session.query(
        User.username,
        User.age,
        subq.label('avg_age')
    ).all()

[OK] SOUS-REQUÊTE DANS FROM
    subq = session.query(
        User.age,
        func.count(User.id).label('count')
    ).group_by(User.age).subquery()
    
    results = session.query(subq).filter(subq.c.count > 5).all()

[OK] EXISTS
    from sqlalchemy import exists
    
    # Vérifier l'existence
    stmt = exists().where(User.username == 'alice')
    has_alice = session.query(stmt).scalar()
    
    # Dans un filtre
    users = session.query(User).filter(
        exists().where(Post.user_id == User.id)
    ).all()


[OK] CHARGEMENT OPTIMISÉ (Eager Loading)


[OK] LAZY LOADING (Par défaut)
    # Les relations sont chargées à la demande (N+1 problem)
    user = session.query(User).first()
    posts = user.posts  # Requête supplémentaire

[OK] JOINED LOAD
    # Charge la relation avec une jointure
    from sqlalchemy.orm import joinedload
    
    users = session.query(User).options(joinedload(User.posts)).all()

[OK] SUBQUERY LOAD
    # Charge la relation avec une sous-requête
    from sqlalchemy.orm import subqueryload
    
    users = session.query(User).options(subqueryload(User.posts)).all()

[OK] SELECT IN LOAD
    # Charge avec SELECT ... IN (recommandé)
    from sqlalchemy.orm import selectinload
    
    users = session.query(User).options(selectinload(User.posts)).all()

[OK] CHARGEMENT MULTIPLE NIVEAUX
    users = session.query(User).options(
        selectinload(User.posts).selectinload(Post.comments)
    ).all()

[OK] NOLOAD (Ne pas charger)
    from sqlalchemy.orm import noload
    
    users = session.query(User).options(noload(User.posts)).all()


[OK] PROPRIÉTÉS HYBRIDES


[OK] HYBRID PROPERTY
    from sqlalchemy.ext.hybrid import hybrid_property
    
    class User(Base):
        __tablename__ = 'users'
        
        id = Column(Integer, primary_key=True)
        first_name = Column(String(50))
        last_name = Column(String(50))
        
        @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)
    
    # Utilisation
    user = session.query(User).first()
    print(user.full_name)  # Instance
    
    # Dans une requête
    users = session.query(User).filter(User.full_name == 'John Doe').all()

[OK] HYBRID METHOD
    from sqlalchemy.ext.hybrid import hybrid_method
    
    class User(Base):
        __tablename__ = 'users'
        
        id = Column(Integer, primary_key=True)
        created_at = Column(DateTime)
        
        @hybrid_method
        def age_in_years(self, reference_date=None):
            if reference_date is None:
                reference_date = datetime.now()
            return (reference_date - self.created_at).days // 365
        
        @age_in_years.expression
        def age_in_years(cls, reference_date=None):
            if reference_date is None:
                reference_date = func.now()
            return func.extract('year', reference_date - cls.created_at)


[OK] TRANSACTIONS


[OK] TRANSACTION BASIQUE
    try:
        user = User(username='test')
        session.add(user)
        session.commit()
    except Exception as e:
        session.rollback()
        print(f"Erreur: {e}")
    finally:
        session.close()

[OK] TRANSACTION AVEC CONTEXT MANAGER
    with session.begin():
        user = User(username='test')
        session.add(user)
        # Commit automatique ou rollback en cas d'erreur

[OK] NESTED TRANSACTION (Savepoint)
    session.begin_nested()
    try:
        user = User(username='test')
        session.add(user)
        session.commit()
    except:
        session.rollback()

[OK] TWO-PHASE COMMIT
    session.begin_twophase()
    try:
        # Opérations
        session.prepare()
        session.commit()
    except:
        session.rollback()


[OK] RAW SQL


[OK] EXÉCUTER DU SQL BRUT
    from sqlalchemy import text
    
    # Simple
    result = session.execute(text("SELECT * FROM users WHERE age > 25"))
    users = result.fetchall()
    
    # Avec paramètres
    result = session.execute(
        text("SELECT * FROM users WHERE age > :age"),
        {"age": 25}
    )
    users = result.fetchall()
    
    # Mapper vers un modèle ORM
    result = session.execute(text("SELECT * FROM users")).mappings()
    for row in result:
        print(row['username'])

[OK] CRÉER DES TABLES AVEC SQL BRUT
    session.execute(text("""
        CREATE TABLE IF NOT EXISTS custom_table (
            id INTEGER PRIMARY KEY,
            name VARCHAR(100)
        )
    """))
    session.commit()


[OK] ÉVÉNEMENTS (Events)


[OK] BEFORE INSERT
    from sqlalchemy import event
    
    @event.listens_for(User, 'before_insert')
    def receive_before_insert(mapper, connection, target):
        print(f"Insertion de: {target.username}")
        target.created_at = datetime.utcnow()

[OK] AFTER INSERT
    @event.listens_for(User, 'after_insert')
    def receive_after_insert(mapper, connection, target):
        print(f"Utilisateur {target.id} créé")

[OK] BEFORE UPDATE
    @event.listens_for(User, 'before_update')
    def receive_before_update(mapper, connection, target):
        target.updated_at = datetime.utcnow()

[OK] BEFORE DELETE
    @event.listens_for(User, 'before_delete')
    def receive_before_delete(mapper, connection, target):
        print(f"Suppression de: {target.username}")

[OK] ÉVÉNEMENTS DE SESSION
    @event.listens_for(Session, 'after_commit')
    def receive_after_commit(session):
        print("Transaction validée")
    
    @event.listens_for(Session, 'after_rollback')
    def receive_after_rollback(session):
        print("Transaction annulée")


[OK] GESTION DES TABLES


[OK] CRÉER TOUTES LES TABLES
    Base.metadata.create_all(engine)

[OK] CRÉER UNE TABLE SPÉCIFIQUE
    User.__table__.create(engine, checkfirst=True)

[OK] SUPPRIMER TOUTES LES TABLES
    Base.metadata.drop_all(engine)

[OK] SUPPRIMER UNE TABLE SPÉCIFIQUE
    User.__table__.drop(engine, checkfirst=True)

[OK] VÉRIFIER SI UNE TABLE EXISTE
    from sqlalchemy import inspect
    
    inspector = inspect(engine)
    table_exists = 'users' in inspector.get_table_names()

[OK] OBTENIR LA STRUCTURE D'UNE TABLE
    inspector = inspect(engine)
    columns = inspector.get_columns('users')
    for col in columns:
        print(f"{col['name']}: {col['type']}")

[OK] RÉFLÉCHIR UNE TABLE EXISTANTE
    from sqlalchemy import MetaData, Table
    
    metadata = MetaData()
    existing_table = Table('users', metadata, autoload_with=engine)
    
    # Accéder aux colonnes
    for column in existing_table.columns:
        print(column.name)


[OK] VERROUILLAGE (Locking)


[OK] FOR UPDATE (Verrouillage exclusif)
    # Verrouiller les lignes pour mise à jour
    user = session.query(User).filter(User.id == 1).with_for_update().first()
    user.balance += 100
    session.commit()

[OK] FOR UPDATE NOWAIT
    # Échoue immédiatement si verrouillé
    try:
        user = session.query(User).filter(User.id == 1).with_for_update(nowait=True).first()
    except Exception as e:
        print("Ligne verrouillée")

[OK] FOR UPDATE SKIP LOCKED
    # Ignore les lignes verrouillées
    users = session.query(User).with_for_update(skip_locked=True).all()

[OK] FOR SHARE (Verrouillage partagé)
    # Permet la lecture mais empêche l'écriture
    user = session.query(User).filter(User.id == 1).with_for_update(read=True).first()


[OK] SERIALIZATION (JSON)


[OK] CONVERTIR EN DICTIONNAIRE
    class User(Base):
        __tablename__ = 'users'
        
        id = Column(Integer, primary_key=True)
        username = Column(String(50))
        email = Column(String(100))
        
        def to_dict(self):
            return {
                'id': self.id,
                'username': self.username,
                'email': self.email
            }
        
        def to_dict_with_relations(self):
            return {
                'id': self.id,
                'username': self.username,
                'posts': [post.to_dict() for post in self.posts]
            }

[OK] AVEC DATACLASS (Python 3.7+)
    from dataclasses import dataclass, asdict
    
    @dataclass
    class UserData:
        id: int
        username: str
        email: str
    
    # Conversion
    user = session.query(User).first()
    user_data = UserData(id=user.id, username=user.username, email=user.email)
    user_dict = asdict(user_data)

[OK] AVEC PYDANTIC
    from pydantic import BaseModel
    
    class UserSchema(BaseModel):
        id: int
        username: str
        email: str
        
        class Config:
            orm_mode = True
    
    # Conversion
    user = session.query(User).first()
    user_schema = UserSchema.from_orm(user)
    user_json = user_schema.json()


[OK] MIGRATIONS AVEC ALEMBIC


[OK] INITIALISER ALEMBIC
    # Dans le terminal
    alembic init alembic

[OK] CONFIGURATION (alembic.ini)
    sqlalchemy.url = postgresql://user:password@localhost/dbname

[OK] CRÉER UNE MIGRATION
    alembic revision --autogenerate -m "Add users table"

[OK] APPLIQUER LES MIGRATIONS
    # Dernière version
    alembic upgrade head
    
    # Version spécifique
    alembic upgrade ae1027a6acf
    
    # Prochaine version
    alembic upgrade +1

[OK] REVENIR EN ARRIÈRE
    # Version précédente
    alembic downgrade -1
    
    # Version spécifique
    alembic downgrade ae1027a6acf
    
    # Tout annuler
    alembic downgrade base

[OK] HISTORIQUE
    alembic history
    alembic current

[OK] MIGRATION MANUELLE
    """Add column to users
    
    Revision ID: xxxxx
    """
    from alembic import op
    import sqlalchemy as sa
    
    def upgrade():
        op.add_column('users', sa.Column('phone', sa.String(20)))
        op.create_index('idx_users_phone', 'users', ['phone'])
    
    def downgrade():
        op.drop_index('idx_users_phone', 'users')
        op.drop_column('users', 'phone')


[OK] PERFORMANCE ET OPTIMISATION


[OK] BATCH OPERATIONS
    # Insert batch
    users = [User(username=f'user{i}') for i in range(1000)]
    session.bulk_save_objects(users)
    session.commit()
    
    # Update batch
    session.bulk_update_mappings(User, [
        {'id': 1, 'age': 25},
        {'id': 2, 'age': 30}
    ])

[OK] DÉSACTIVER AUTOFLUSH TEMPORAIREMENT
    with session.no_autoflush:
        # Opérations sans flush automatique
        user = session.query(User).first()
        user.age = 30

[OK] EXPIRE ALL (Rafraîchir tous les objets)
    session.expire_all()

[OK] EXPUNGE (Retirer de la session)
    user = session.query(User).first()
    session.expunge(user)
    # user n'est plus tracké par la session

[OK] MERGE (Rattacher un objet détaché)
    detached_user = User(id=1, username='alice')
    attached_user = session.merge(detached_user)

[OK] QUERY OPTIONS POUR PERFORMANCE
    from sqlalchemy.orm import lazyload, raiseload
    
    # Désactiver le lazy loading (erreur si accès)
    users = session.query(User).options(raiseload('*')).all()
    
    # Forcer le lazy loading
    users = session.query(User).options(lazyload(User.posts)).all()

[OK] COMPILATION DE REQUÊTES
    # Compiler et afficher la requête SQL
    query = session.query(User).filter(User.age > 25)
    print(str(query.statement.compile(compile_kwargs={"literal_binds": True})))

[OK] INDEXATION
    class User(Base):
        __tablename__ = 'users'
        
        id = Column(Integer, primary_key=True)
        email = Column(String(100))
        username = Column(String(50))
        
        __table_args__ = (
            Index('idx_email', 'email'),
            Index('idx_username_email', 'username', 'email'),
            Index('idx_email_partial', 'email', postgresql_where=Column('is_active') == True),
        )


[OK] TESTING


[OK] BASE DE DONNÉES EN MÉMOIRE POUR TESTS
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    
    # Engine de test
    test_engine = create_engine('sqlite:///:memory:')
    TestSession = sessionmaker(bind=test_engine)
    
    # Créer les tables
    Base.metadata.create_all(test_engine)
    
    # Utiliser dans les tests
    session = TestSession()
    # ... vos tests
    session.close()

[OK] FIXTURE PYTEST
    import pytest
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    
    @pytest.fixture(scope='function')
    def db_session():
        engine = create_engine('sqlite:///:memory:')
        Base.metadata.create_all(engine)
        Session = sessionmaker(bind=engine)
        session = Session()
        yield session
        session.close()
    
    # Utilisation
    def test_user_creation(db_session):
        user = User(username='test')
        db_session.add(user)
        db_session.commit()
        assert user.id is not None

[OK] ROLLBACK APRÈS CHAQUE TEST
    @pytest.fixture(scope='function')
    def db_session():
        connection = engine.connect()
        transaction = connection.begin()
        session = Session(bind=connection)
        
        yield session
        
        session.close()
        transaction.rollback()
        connection.close()


[OK] SQLALCHEMY ASYNC (SQLAlchemy 2.0+)


[OK] CONFIGURATION ASYNC
    from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
    from sqlalchemy.ext.asyncio import async_sessionmaker
    
    # Créer un engine async
    async_engine = create_async_engine(
        'postgresql+asyncpg://user:pass@localhost/dbname',
        echo=True
    )
    
    # Session factory
    async_session = async_sessionmaker(async_engine, expire_on_commit=False)

[OK] REQUÊTES ASYNC
    from sqlalchemy import select
    
    async def get_users():
        async with async_session() as session:
            stmt = select(User).where(User.age > 25)
            result = await session.execute(stmt)
            users = result.scalars().all()
            return users

[OK] INSERT ASYNC
    async def create_user(username, email):
        async with async_session() as session:
            user = User(username=username, email=email)
            session.add(user)
            await session.commit()
            return user

[OK] UPDATE ASYNC
    async def update_user_age(user_id, new_age):
        async with async_session() as session:
            stmt = select(User).where(User.id == user_id)
            result = await session.execute(stmt)
            user = result.scalar_one()
            user.age = new_age
            await session.commit()

[OK] DELETE ASYNC
    async def delete_user(user_id):
        async with async_session() as session:
            stmt = select(User).where(User.id == user_id)
            result = await session.execute(stmt)
            user = result.scalar_one()
            await session.delete(user)
            await session.commit()

[OK] EAGER LOADING ASYNC
    from sqlalchemy.orm import selectinload
    
    async def get_users_with_posts():
        async with async_session() as session:
            stmt = select(User).options(selectinload(User.posts))
            result = await session.execute(stmt)
            users = result.scalars().all()
            return users


[OK] SQLALCHEMY 2.0 STYLE (Nouveau)


[OK] SELECT AVEC SELECT()
    from sqlalchemy import select
    
    # Nouveau style (recommandé)
    stmt = select(User).where(User.age > 25)
    result = session.execute(stmt)
    users = result.scalars().all()
    
    # Ancien style (déprécié)
    users = session.query(User).filter(User.age > 25).all()

[OK] INSERT AVEC INSERT()
    from sqlalchemy import insert
    
    stmt = insert(User).values(username='alice', email='alice@example.com')
    session.execute(stmt)
    session.commit()

[OK] UPDATE AVEC UPDATE()
    from sqlalchemy import update
    
    stmt = update(User).where(User.id == 1).values(age=30)
    session.execute(stmt)
    session.commit()

[OK] DELETE AVEC DELETE()
    from sqlalchemy import delete
    
    stmt = delete(User).where(User.age < 18)
    session.execute(stmt)
    session.commit()

[OK] SCALARS() POUR RÉSULTATS
    # Obtenir une liste d'objets
    stmt = select(User)
    result = session.execute(stmt)
    users = result.scalars().all()
    
    # Obtenir un seul objet
    user = result.scalars().first()
    user = result.scalar_one()  # Erreur si 0 ou >1
    user = result.scalar_one_or_none()  # None si 0, erreur si >1


[OK] UTILITAIRES ET ASTUCES


[OK] OBTENIR LE SQL GÉNÉRÉ
    query = session.query(User).filter(User.age > 25)
    print(query)
    
    # Avec les valeurs liées
    from sqlalchemy.dialects import postgresql
    print(query.statement.compile(
        dialect=postgresql.dialect(),
        compile_kwargs={"literal_binds": True}
    ))

[OK] VÉRIFIER SI UN OBJET EST DANS LA SESSION
    user = User(username='test')
    print(user in session)  # False
    session.add(user)
    print(user in session)  # True

[OK] OBTENIR L'ÉTAT D'UN OBJET
    from sqlalchemy import inspect
    
    user = session.query(User).first()
    state = inspect(user)
    
    print(state.persistent)  # Dans la session et la DB
    print(state.pending)     # Dans la session, pas encore en DB
    print(state.transient)   # Pas dans la session
    print(state.detached)    # Plus dans la session mais était en DB

[OK] CLONER UN OBJET
    from sqlalchemy.orm import make_transient
    
    user = session.query(User).first()
    session.expunge(user)
    make_transient(user)
    user.id = None  # Nouveau ID sera généré
    session.add(user)
    session.commit()

[OK] PAGINATION HELPER
    def paginate(query, page, per_page=20):
        items = query.limit(per_page).offset((page - 1) * per_page).all()
        total = query.count()
        pages = (total + per_page - 1) // per_page
        return {
            'items': items,
            'total': total,
            'page': page,
            'per_page': per_page,
            'pages': pages,
            'has_prev': page > 1,
            'has_next': page < pages
        }
    
    # Utilisation
    result = paginate(session.query(User), page=1, per_page=10)

[OK] LOGGER LES REQUÊTES
    import logging
    
    logging.basicConfig()
    logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

[OK] DÉSACTIVER TEMPORAIREMENT L'ECHO
    with engine.connect() as conn:
        with conn.execution_options(logging_token="special"):
            # Vos requêtes ici
            pass

[OK] CRÉER UN MIXIN POUR COLONNES COMMUNES
    from datetime import datetime
    from sqlalchemy import Column, Integer, DateTime
    
    class TimestampMixin:
        created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
        updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    class User(Base, TimestampMixin):
        __tablename__ = 'users'
        id = Column(Integer, primary_key=True)
        username = Column(String(50))

[OK] VALIDATION PERSONNALISÉE
    from sqlalchemy.orm import validates
    
    class User(Base):
        __tablename__ = 'users'
        
        id = Column(Integer, primary_key=True)
        email = Column(String(100))
        age = Column(Integer)
        
        @validates('email')
        def validate_email(self, key, email):
            if '@' not in email:
                raise ValueError("Email invalide")
            return email
        
        @validates('age')
        def validate_age(self, key, age):
            if age < 0 or age > 150:
                raise ValueError("Âge invalide")
            return age

[OK] UNION DE REQUÊTES
    from sqlalchemy import union
    
    query1 = session.query(User.username).filter(User.age < 20)
    query2 = session.query(User.username).filter(User.age > 60)
    
    combined = union(query1, query2)
    results = session.execute(combined).all()

[OK] WINDOW FUNCTIONS
    from sqlalchemy import over
    
    # ROW_NUMBER
    stmt = select(
        User.username,
        func.row_number().over(order_by=User.created_at).label('row_num')
    )
    
    # RANK avec PARTITION BY
    stmt = select(
        User.username,
        User.age,
        func.rank().over(
            partition_by=User.age,
            order_by=User.created_at
        ).label('rank')
    )

[OK] CTE (Common Table Expression)
    from sqlalchemy import select
    
    # Créer un CTE
    cte = select(User.age, func.count(User.id).label('count')).group_by(User.age).cte('age_counts')
    
    # Utiliser le CTE
    stmt = select(cte.c.age, cte.c.count).where(cte.c.count > 5)
    result = session.execute(stmt).all()


[OK] BONNES PRATIQUES


[OK] TOUJOURS FERMER LES SESSIONS
    # Méthode 1: try/finally
    session = Session()
    try:
        # opérations
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()
    
    # Méthode 2: context manager (recommandé)
    with Session() as session:
        # opérations
        session.commit()

[OK] UTILISER DES INDEXES APPROPRIÉS
    # Index sur colonnes fréquemment filtrées
    # Index composites pour requêtes multi-colonnes
    # Éviter trop d'indexes (ralentit les écritures)

[OK] ÉVITER LE N+1 PROBLEM
    # Mauvais: N+1 requêtes
    users = session.query(User).all()
    for user in users:
        print(user.posts)  # Requête par user
    
    # Bon: 1 ou 2 requêtes
    users = session.query(User).options(selectinload(User.posts)).all()
    for user in users:
        print(user.posts)

[OK] UTILISER BULK OPERATIONS POUR GROS VOLUMES
    # Au lieu de add() individuel, utiliser bulk_save_objects()
    # Désactiver autoflush si nécessaire

[OK] DÉFINIR CASCADE APPROPRIÉ
    # delete-orphan pour supprimer les enfants orphelins
    # all pour propager tous les événements

[OK] VALIDATION DES DONNÉES
    # Utiliser @validates pour validation au niveau ORM
    # Utiliser CheckConstraint pour validation en base
    # Combiner les deux pour robustesse maximale

[OK] TRANSACTIONS EXPLICITES
    # Toujours wrapper les opérations critiques dans des transactions
    # Utiliser SAVEPOINT pour transactions imbriquées


[OK] RESSOURCES


Documentation officielle: https://docs.sqlalchemy.org/
Tutorial SQLAlchemy 2.0: https://docs.sqlalchemy.org/en/20/tutorial/
ORM Querying Guide: https://docs.sqlalchemy.org/en/20/orm/queryguide/
SQLAlchemy sur GitHub: https://github.com/sqlalchemy/sqlalchemy


# FIN DU CHEATSHEET
