# Fichier: python_cheats/cheatsheets/database.txt
# Cheatsheet Bases de Données Python - Guide Complet


[OK] SQLITE3 (STANDARD LIBRARY)

# === CONNEXION ===

import sqlite3

# Créer/Ouvrir base de données
conn = sqlite3.connect('database.db')
conn = sqlite3.connect('data/mydb.sqlite')
conn = sqlite3.connect(':memory:')              # Base en mémoire

# Options de connexion
conn = sqlite3.connect('database.db', 
    timeout=10.0,                               # Timeout en secondes
    isolation_level='DEFERRED',                 # DEFERRED, IMMEDIATE, EXCLUSIVE
    check_same_thread=False                     # Multi-threading
)

# Créer curseur
cursor = conn.cursor()

# Fermer connexion
cursor.close()
conn.close()

# Context manager (recommandé)
with sqlite3.connect('database.db') as conn:
    cursor = conn.cursor()
    # Opérations
    # Auto-commit et close

# === CRÉATION DE TABLES ===

# Créer table
cursor.execute('''
    CREATE TABLE users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        username TEXT NOT NULL UNIQUE,
        email TEXT NOT NULL,
        age INTEGER,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

# Créer table si n'existe pas
cursor.execute('''
    CREATE TABLE IF NOT EXISTS products (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        price REAL,
        stock INTEGER DEFAULT 0
    )
''')

# Table avec clés étrangères
cursor.execute('''
    CREATE TABLE orders (
        id INTEGER PRIMARY KEY,
        user_id INTEGER NOT NULL,
        product_id INTEGER NOT NULL,
        quantity INTEGER,
        FOREIGN KEY (user_id) REFERENCES users(id),
        FOREIGN KEY (product_id) REFERENCES products(id)
    )
''')

# Activer clés étrangères
cursor.execute('PRAGMA foreign_keys = ON')

# Table avec index
cursor.execute('CREATE INDEX idx_username ON users(username)')
cursor.execute('CREATE UNIQUE INDEX idx_email ON users(email)')

# === INSERTION ===

# Insertion simple
cursor.execute(
    'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
    ('john_doe', 'john@example.com', 30)
)

# Insertion avec dictionnaire
cursor.execute(
    'INSERT INTO users (username, email) VALUES (:username, :email)',
    {'username': 'jane_doe', 'email': 'jane@example.com'}
)

# Insertions multiples
users = [
    ('alice', 'alice@example.com', 25),
    ('bob', 'bob@example.com', 35),
    ('charlie', 'charlie@example.com', 28)
]
cursor.executemany(
    'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
    users
)

# Récupérer ID du dernier insert
last_id = cursor.lastrowid
print(f"ID inséré: {last_id}")

# INSERT OR IGNORE
cursor.execute(
    'INSERT OR IGNORE INTO users (username, email) VALUES (?, ?)',
    ('existing_user', 'email@example.com')
)

# INSERT OR REPLACE
cursor.execute(
    'INSERT OR REPLACE INTO users (id, username, email) VALUES (?, ?, ?)',
    (1, 'updated_user', 'new@example.com')
)

# Commit
conn.commit()

# === SÉLECTION ===

# Sélectionner tout
cursor.execute('SELECT * FROM users')
rows = cursor.fetchall()
for row in rows:
    print(row)

# Sélectionner un seul résultat
cursor.execute('SELECT * FROM users WHERE id = ?', (1,))
user = cursor.fetchone()
print(user)

# Sélectionner avec LIMIT
cursor.execute('SELECT * FROM users LIMIT 10')
rows = cursor.fetchmany(10)

# Sélectionner colonnes spécifiques
cursor.execute('SELECT username, email FROM users')

# WHERE conditions
cursor.execute('SELECT * FROM users WHERE age > ?', (25,))
cursor.execute('SELECT * FROM users WHERE username LIKE ?', ('%john%',))

# ORDER BY
cursor.execute('SELECT * FROM users ORDER BY age DESC')

# JOIN
cursor.execute('''
    SELECT users.username, orders.quantity, products.name
    FROM orders
    JOIN users ON orders.user_id = users.id
    JOIN products ON orders.product_id = products.id
''')

# GROUP BY
cursor.execute('''
    SELECT user_id, COUNT(*) as order_count
    FROM orders
    GROUP BY user_id
''')

# Itérer sur résultats
cursor.execute('SELECT * FROM users')
for row in cursor:
    print(row)

# Compter lignes
cursor.execute('SELECT COUNT(*) FROM users')
count = cursor.fetchone()[0]

# === MISE À JOUR ===

# UPDATE simple
cursor.execute(
    'UPDATE users SET age = ? WHERE username = ?',
    (31, 'john_doe')
)

# UPDATE multiple colonnes
cursor.execute(
    'UPDATE users SET email = ?, age = ? WHERE id = ?',
    ('newemail@example.com', 35, 1)
)

# Nombre de lignes affectées
rows_affected = cursor.rowcount
print(f"{rows_affected} lignes mises à jour")

conn.commit()

# === SUPPRESSION ===

# DELETE
cursor.execute('DELETE FROM users WHERE id = ?', (5,))

# DELETE avec condition
cursor.execute('DELETE FROM users WHERE age < ?', (18,))

# DELETE all
cursor.execute('DELETE FROM users')

# TRUNCATE (via DELETE)
cursor.execute('DELETE FROM users')
cursor.execute('VACUUM')  # Récupérer espace

conn.commit()

# === TRANSACTIONS ===

try:
    conn.execute('BEGIN')
    cursor.execute('INSERT INTO users (username, email) VALUES (?, ?)', 
                   ('test', 'test@example.com'))
    cursor.execute('UPDATE products SET stock = stock - 1 WHERE id = ?', (1,))
    conn.commit()
except Exception as e:
    conn.rollback()
    print(f"Erreur: {e}")

# Context manager avec transaction
with conn:
    cursor.execute('INSERT INTO users (username, email) VALUES (?, ?)',
                   ('auto_commit', 'auto@example.com'))
    # Auto-commit si pas d'exception

# === ROW FACTORY ===

# Accéder colonnes par nom
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('SELECT * FROM users WHERE id = ?', (1,))
row = cursor.fetchone()
print(row['username'])
print(row['email'])

# Convertir Row en dict
user_dict = dict(row)

# Row factory personnalisé
def dict_factory(cursor, row):
    return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}

conn.row_factory = dict_factory

# === FONCTIONS UTILES ===

# Vérifier si table existe
cursor.execute("""
    SELECT name FROM sqlite_master 
    WHERE type='table' AND name='users'
""")
exists = cursor.fetchone() is not None

# Lister toutes les tables
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()

# Obtenir schéma de table
cursor.execute("PRAGMA table_info(users)")
columns = cursor.fetchall()

# Supprimer table
cursor.execute('DROP TABLE IF EXISTS users')

# Renommer table
cursor.execute('ALTER TABLE users RENAME TO customers')

# Ajouter colonne
cursor.execute('ALTER TABLE users ADD COLUMN phone TEXT')

# Sauvegarder base de données
def backup_database(source_db, dest_db):
    source = sqlite3.connect(source_db)
    dest = sqlite3.connect(dest_db)
    source.backup(dest)
    dest.close()
    source.close()

# === EXEMPLE COMPLET ===

import sqlite3
from datetime import datetime

class Database:
    def __init__(self, db_name):
        self.conn = sqlite3.connect(db_name)
        self.conn.row_factory = sqlite3.Row
        self.cursor = self.conn.cursor()
        self.create_tables()
    
    def create_tables(self):
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT NOT NULL UNIQUE,
                email TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.conn.commit()
    
    def add_user(self, username, email):
        try:
            self.cursor.execute(
                'INSERT INTO users (username, email) VALUES (?, ?)',
                (username, email)
            )
            self.conn.commit()
            return self.cursor.lastrowid
        except sqlite3.IntegrityError:
            return None
    
    def get_user(self, user_id):
        self.cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
        return self.cursor.fetchone()
    
    def get_all_users(self):
        self.cursor.execute('SELECT * FROM users')
        return self.cursor.fetchall()
    
    def update_user(self, user_id, username=None, email=None):
        if username:
            self.cursor.execute('UPDATE users SET username = ? WHERE id = ?',
                                (username, user_id))
        if email:
            self.cursor.execute('UPDATE users SET email = ? WHERE id = ?',
                                (email, user_id))
        self.conn.commit()
    
    def delete_user(self, user_id):
        self.cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
        self.conn.commit()
    
    def close(self):
        self.conn.close()


[OK] MYSQL (MYSQL-CONNECTOR-PYTHON)

# === INSTALLATION ===
# pip install mysql-connector-python

import mysql.connector
from mysql.connector import Error

# === CONNEXION ===

# Connexion simple
conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='mydb'
)

# Connexion avec options
conn = mysql.connector.connect(
    host='localhost',
    port=3306,
    user='root',
    password='password',
    database='mydb',
    charset='utf8mb4',
    collation='utf8mb4_unicode_ci',
    autocommit=False,
    pool_name='mypool',
    pool_size=5
)

# Vérifier connexion
if conn.is_connected():
    print("Connecté à MySQL")

# Context manager
with mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='mydb'
) as conn:
    cursor = conn.cursor()
    # Opérations

# === CRÉATION DE BASE ===

# Créer base de données
cursor = conn.cursor()
cursor.execute('CREATE DATABASE IF NOT EXISTS mydb')
cursor.execute('USE mydb')

# === CRÉATION DE TABLES ===

cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INT AUTO_INCREMENT PRIMARY KEY,
        username VARCHAR(50) NOT NULL UNIQUE,
        email VARCHAR(100) NOT NULL,
        age INT,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        INDEX idx_username (username)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
''')

# Table avec clé étrangère
cursor.execute('''
    CREATE TABLE IF NOT EXISTS orders (
        id INT AUTO_INCREMENT PRIMARY KEY,
        user_id INT NOT NULL,
        product_name VARCHAR(100),
        quantity INT,
        FOREIGN KEY (user_id) REFERENCES users(id)
            ON DELETE CASCADE
            ON UPDATE CASCADE
    ) ENGINE=InnoDB
''')

# === INSERTION ===

# Insertion simple
query = 'INSERT INTO users (username, email, age) VALUES (%s, %s, %s)'
values = ('john_doe', 'john@example.com', 30)
cursor.execute(query, values)
conn.commit()

# ID du dernier insert
last_id = cursor.lastrowid

# Insertions multiples
query = 'INSERT INTO users (username, email, age) VALUES (%s, %s, %s)'
users = [
    ('alice', 'alice@example.com', 25),
    ('bob', 'bob@example.com', 35),
    ('charlie', 'charlie@example.com', 28)
]
cursor.executemany(query, users)
conn.commit()

# === SÉLECTION ===

# Sélection simple
cursor.execute('SELECT * FROM users')
results = cursor.fetchall()
for row in results:
    print(row)

# Sélection avec WHERE
cursor.execute('SELECT * FROM users WHERE age > %s', (25,))
results = cursor.fetchall()

# Fetchone
cursor.execute('SELECT * FROM users WHERE id = %s', (1,))
user = cursor.fetchone()

# Curseur dictionnaire
cursor = conn.cursor(dictionary=True)
cursor.execute('SELECT * FROM users')
users = cursor.fetchall()
for user in users:
    print(user['username'], user['email'])

# === MISE À JOUR ===

query = 'UPDATE users SET age = %s WHERE username = %s'
cursor.execute(query, (31, 'john_doe'))
conn.commit()
print(f"{cursor.rowcount} lignes mises à jour")

# === SUPPRESSION ===

cursor.execute('DELETE FROM users WHERE id = %s', (5,))
conn.commit()

# === TRANSACTIONS ===

try:
    conn.start_transaction()
    cursor.execute('INSERT INTO users (username, email) VALUES (%s, %s)',
                   ('test', 'test@example.com'))
    cursor.execute('UPDATE orders SET quantity = quantity - 1 WHERE id = %s', (1,))
    conn.commit()
except Error as e:
    conn.rollback()
    print(f"Erreur: {e}")

# === PROCÉDURES STOCKÉES ===

# Appeler procédure
cursor.callproc('get_user_orders', [1])
for result in cursor.stored_results():
    results = result.fetchall()

# === FERMETURE ===

cursor.close()
conn.close()

# === EXEMPLE CLASSE ===

class MySQLDatabase:
    def __init__(self, host, user, password, database):
        self.config = {
            'host': host,
            'user': user,
            'password': password,
            'database': database
        }
        self.conn = None
        self.cursor = None
    
    def connect(self):
        try:
            self.conn = mysql.connector.connect(**self.config)
            self.cursor = self.conn.cursor(dictionary=True)
            return True
        except Error as e:
            print(f"Erreur connexion: {e}")
            return False
    
    def execute_query(self, query, params=None):
        try:
            self.cursor.execute(query, params or ())
            self.conn.commit()
            return True
        except Error as e:
            print(f"Erreur: {e}")
            self.conn.rollback()
            return False
    
    def fetch_all(self, query, params=None):
        try:
            self.cursor.execute(query, params or ())
            return self.cursor.fetchall()
        except Error as e:
            print(f"Erreur: {e}")
            return []
    
    def close(self):
        if self.cursor:
            self.cursor.close()
        if self.conn:
            self.conn.close()


[OK] POSTGRESQL (PSYCOPG2)

# === INSTALLATION ===
# pip install psycopg2-binary

import psycopg2
from psycopg2 import sql, Error
from psycopg2.extras import RealDictCursor, execute_values

# === CONNEXION ===

# Connexion simple
conn = psycopg2.connect(
    host='localhost',
    port=5432,
    database='mydb',
    user='postgres',
    password='password'
)

# Connection string
conn = psycopg2.connect('postgresql://user:password@localhost:5432/mydb')

# Connexion avec options
conn = psycopg2.connect(
    host='localhost',
    database='mydb',
    user='postgres',
    password='password',
    connect_timeout=10,
    options='-c search_path=public'
)

# Autocommit
conn.autocommit = True

# === CURSEURS ===

# Curseur standard
cursor = conn.cursor()

# Curseur dictionnaire
cursor = conn.cursor(cursor_factory=RealDictCursor)

# === CRÉATION DE TABLES ===

cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id SERIAL PRIMARY KEY,
        username VARCHAR(50) NOT NULL UNIQUE,
        email VARCHAR(100) NOT NULL,
        age INTEGER,
        is_active BOOLEAN DEFAULT TRUE,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')
conn.commit()

# Table avec contraintes
cursor.execute('''
    CREATE TABLE IF NOT EXISTS products (
        id SERIAL PRIMARY KEY,
        name VARCHAR(100) NOT NULL,
        price DECIMAL(10, 2) CHECK (price >= 0),
        stock INTEGER DEFAULT 0,
        category VARCHAR(50),
        created_at TIMESTAMP DEFAULT NOW()
    )
''')

# Index
cursor.execute('CREATE INDEX idx_username ON users(username)')
cursor.execute('CREATE INDEX idx_email ON users USING btree(email)')

# === INSERTION ===

# Insertion simple
cursor.execute(
    'INSERT INTO users (username, email, age) VALUES (%s, %s, %s)',
    ('john_doe', 'john@example.com', 30)
)
conn.commit()

# RETURNING clause
cursor.execute(
    'INSERT INTO users (username, email) VALUES (%s, %s) RETURNING id',
    ('jane_doe', 'jane@example.com')
)
user_id = cursor.fetchone()[0]

# Insertion avec dictionnaire
cursor.execute(
    'INSERT INTO users (username, email) VALUES (%(username)s, %(email)s)',
    {'username': 'bob', 'email': 'bob@example.com'}
)

# Insertions multiples optimisées
users = [
    ('alice', 'alice@example.com', 25),
    ('bob', 'bob@example.com', 35),
    ('charlie', 'charlie@example.com', 28)
]
execute_values(
    cursor,
    'INSERT INTO users (username, email, age) VALUES %s',
    users
)
conn.commit()

# === SÉLECTION ===

# Sélection avec RealDictCursor
cursor = conn.cursor(cursor_factory=RealDictCursor)
cursor.execute('SELECT * FROM users')
users = cursor.fetchall()
for user in users:
    print(user['username'], user['email'])

# WHERE avec paramètres
cursor.execute('SELECT * FROM users WHERE age > %s', (25,))

# LIKE
cursor.execute('SELECT * FROM users WHERE username LIKE %s', ('%john%',))

# IN clause
cursor.execute('SELECT * FROM users WHERE id IN %s', ([1, 2, 3],))

# JSON query (PostgreSQL)
cursor.execute('''
    SELECT * FROM users 
    WHERE metadata @> '{"active": true}'::jsonb
''')

# === MISE À JOUR ===

cursor.execute(
    'UPDATE users SET age = %s WHERE username = %s',
    (31, 'john_doe')
)
conn.commit()

# UPDATE avec RETURNING
cursor.execute(
    'UPDATE users SET age = %s WHERE id = %s RETURNING *',
    (35, 1)
)
updated_user = cursor.fetchone()

# === SUPPRESSION ===

cursor.execute('DELETE FROM users WHERE id = %s', (5,))
conn.commit()

# DELETE avec RETURNING
cursor.execute('DELETE FROM users WHERE age < %s RETURNING id', (18,))
deleted_ids = cursor.fetchall()

# === TRANSACTIONS ===

try:
    cursor.execute('BEGIN')
    cursor.execute('INSERT INTO users (username, email) VALUES (%s, %s)',
                   ('test', 'test@example.com'))
    cursor.execute('UPDATE products SET stock = stock - 1 WHERE id = %s', (1,))
    conn.commit()
except Exception as e:
    conn.rollback()
    print(f"Erreur: {e}")

# Context manager
with conn:
    with conn.cursor() as cursor:
        cursor.execute('INSERT INTO users (username, email) VALUES (%s, %s)',
                       ('auto', 'auto@example.com'))
    # Auto-commit

# === TYPES SPÉCIAUX POSTGRESQL ===

# Array
cursor.execute(
    'INSERT INTO table (tags) VALUES (%s)',
    ([['python', 'database', 'tutorial']],)
)

# JSON/JSONB
import json
data = {'name': 'John', 'age': 30}
cursor.execute(
    'INSERT INTO table (data) VALUES (%s)',
    (json.dumps(data),)
)

# UUID
import uuid
cursor.execute(
    'INSERT INTO table (id) VALUES (%s)',
    (uuid.uuid4(),)
)

# === COPY (IMPORT RAPIDE) ===

# Importer depuis CSV
with open('users.csv', 'r') as f:
    cursor.copy_from(f, 'users', sep=',', columns=('username', 'email', 'age'))
conn.commit()

# Exporter vers CSV
with open('users_export.csv', 'w') as f:
    cursor.copy_to(f, 'users', sep=',')

# === EXEMPLE CLASSE ===

class PostgreSQLDatabase:
    def __init__(self, host, database, user, password):
        self.conn = psycopg2.connect(
            host=host,
            database=database,
            user=user,
            password=password
        )
        self.conn.autocommit = True
    
    def execute(self, query, params=None):
        with self.conn.cursor() as cursor:
            cursor.execute(query, params)
            if cursor.description:
                return cursor.fetchall()
    
    def fetch_dict(self, query, params=None):
        with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
            cursor.execute(query, params)
            return cursor.fetchall()
    
    def close(self):
        self.conn.close()


[OK] SQLALCHEMY (ORM)

# === INSTALLATION ===
# pip install sqlalchemy

from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean
from sqlalchemy import ForeignKey, DateTime, Text, Table, MetaData
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, Session
from datetime import datetime

# === CONNEXION ===

# SQLite
engine = create_engine('sqlite:///database.db', echo=True)

# MySQL
engine = create_engine('mysql+mysqlconnector://user:pass@localhost/db')

# PostgreSQL
engine = create_engine('postgresql://user:pass@localhost/db')

# Options
engine = create_engine(
    'sqlite:///database.db',
    echo=True,              # Log SQL
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True      # Vérifier connexion
)

# === DÉFINITION MODÈLES ===

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)
    age = Column(Integer)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relation
    orders = relationship('Order', back_populates='user')
    
    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"

class Product(Base):
    __tablename__ = 'products'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    price = Column(Float)
    stock = Column(Integer, default=0)
    
    orders = relationship('Order', back_populates='product')

class Order(Base):
    __tablename__ = 'orders'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    product_id = Column(Integer, ForeignKey('products.id'))
    quantity = Column(Integer)
    
    user = relationship('User', back_populates='orders')
    product = relationship('Product', back_populates='orders')

# Créer toutes les tables
Base.metadata.create_all(engine)

# === SESSION ===

# Créer session factory
Session = sessionmaker(bind=engine)
session = Session()

# Context manager
from contextlib import contextmanager

@contextmanager
def get_session():
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

# === INSERTION ===

# Créer objet
user = User(username='john_doe', email='john@example.com', age=30)
session.add(user)
session.commit()

# Récupérer ID
print(user.id)

# Ajouter plusieurs
users = [
    User(username='alice', email='alice@example.com', age=25),
    User(username='bob', email='bob@example.com', age=35)
]
session.add_all(users)
session.commit()

# === REQUÊTES ===

# Sélectionner tout
users = session.query(User).all()

# Sélectionner avec filter
user = session.query(User).filter_by(username='john_doe').first()
user = session.query(User).filter(User.username == 'john_doe').first()

# Conditions multiples
users = session.query(User).filter(
    User.age > 25,
    User.is_active == True
).all()

# OR
from sqlalchemy import or_
users = session.query(User).filter(
    or_(User.age < 20, User.age > 60)
).all()

# AND
from sqlalchemy import and_
users = session.query(User).filter(
    and_(User.age > 20, User.age < 40)
).all()

# LIKE
users = session.query(User).filter(User.username.like('%john%')).all()

# IN
users = session.query(User).filter(User.id.in_([1, 2, 3])).all()

# ORDER BY
users = session.query(User).order_by(User.age.desc()).all()

# LIMIT & OFFSET
users = session.query(User).limit(10).offset(20).all()

# COUNT
count = session.query(User).count()

# Colonnes spécifiques
results = session.query(User.username, User.email).all()

# JOIN
results = session.query(User, Order).join(Order).all()

# LEFT JOIN
results = session.query(User).outerjoin(Order).all()

# GROUP BY
from sqlalchemy import func
results = session.query(
    User.age,
    func.count(User.id)
).group_by(User.age).all()

# === MISE À JOUR ===

# Méthode 1: Récupérer et modifier
user = session.query(User).filter_by(id=1).first()
user.age = 31
session.commit()

# Méthode 2: Update direct
session.query(User).filter_by(id=1).update({'age': 31})
session.commit()

# === SUPPRESSION ===

# Méthode 1: Récupérer et supprimer
user = session.query(User).filter_by(id=1).first()
session.delete(user)
session.commit()

# Méthode 2: Delete direct
session.query(User).filter_by(id=1).delete()
session.commit()

# === RELATIONS ===

# Créer avec relation
user = User(username='charlie', email='charlie@example.com')
order = Order(quantity=5)
order.user = user
session.add(order)
session.commit()

# Accéder relation
user = session.query(User).filter_by(id=1).first()
print(user.orders)  # Liste des commandes

# Lazy loading vs Eager loading
from sqlalchemy.orm import joinedload
user = session.query(User).options(joinedload(User.orders)).first()

# === TRANSACTIONS ===

try:
    user = User(username='test', email='test@example.com')
    session.add(user)
    session.flush()  # Exécute sans commit
    # Autres opérations
    session.commit()
except Exception as e:
    session.rollback()
    print(f"Erreur: {e}")

# === EXEMPLE COMPLET ===

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

Base = declarative_base()

class Database:
    def __init__(self, db_url):
        self.engine = create_engine(db_url)
        Base.metadata.create_all(self.engine)
        self.SessionLocal = sessionmaker(bind=self.engine)
    
    def get_session(self) -> Session:
        return self.SessionLocal()
    
    def create_user(self, username, email, age):
        session = self.get_session()
        try:
            user = User(username=username, email=email, age=age)
            session.add(user)
            session.commit()
            session.refresh(user)
            return user
        except Exception as e:
            session.rollback()
            raise e
        finally:
            session.close()
    
    def get_user(self, user_id):
        session = self.get_session()
        try:
            return session.query(User).filter_by(id=user_id).first()
        finally:
            session.close()
    
    def get_all_users(self):
        session = self.get_session()
        try:
            return session.query(User).all()
        finally:
            session.close()


[OK] PYMONGO (MONGODB)

# === INSTALLATION ===
# pip install pymongo

from pymongo import MongoClient
from pymongo.errors import ConnectionFailure, PyMongoError
from bson.objectid import ObjectId
from datetime import datetime

# === CONNEXION ===

# Connexion locale
client = MongoClient('localhost', 27017)
client = MongoClient('mongodb://localhost:27017/')

# Connexion avec authentification
client = MongoClient(
    'mongodb://username:password@localhost:27017/',
    authSource='admin'
)

# MongoDB Atlas (cloud)
client = MongoClient(
    'mongodb+srv://user:pass@cluster.mongodb.net/mydb?retryWrites=true&w=majority'
)

# Options de connexion
client = MongoClient(
    'localhost',
    27017,
    maxPoolSize=50,
    minPoolSize=10,
    serverSelectionTimeoutMS=5000
)

# Vérifier connexion
try:
    client.admin.command('ping')
    print("Connecté à MongoDB")
except ConnectionFailure:
    print("Échec de connexion")

# === BASE DE DONNÉES ===

# Accéder/créer base de données
db = client['mydb']
db = client.mydb

# Lister bases de données
databases = client.list_database_names()

# Supprimer base de données
client.drop_database('mydb')

# === COLLECTIONS ===

# Accéder/créer collection
collection = db['users']
collection = db.users

# Lister collections
collections = db.list_collection_names()

# Créer collection avec options
db.create_collection('users', capped=True, size=1000000, max=100)

# Supprimer collection
db.drop_collection('users')

# === INSERTION ===

# Insérer un document
user = {
    'username': 'john_doe',
    'email': 'john@example.com',
    'age': 30,
    'created_at': datetime.utcnow()
}
result = collection.insert_one(user)
print(f"ID inséré: {result.inserted_id}")

# Insérer plusieurs documents
users = [
    {'username': 'alice', 'email': 'alice@example.com', 'age': 25},
    {'username': 'bob', 'email': 'bob@example.com', 'age': 35},
    {'username': 'charlie', 'email': 'charlie@example.com', 'age': 28}
]
result = collection.insert_many(users)
print(f"IDs insérés: {result.inserted_ids}")

# Document avec sous-documents
user = {
    'username': 'john_doe',
    'email': 'john@example.com',
    'address': {
        'street': '123 Main St',
        'city': 'New York',
        'zip': '10001'
    },
    'tags': ['python', 'mongodb', 'developer']
}
collection.insert_one(user)

# === RECHERCHE ===

# Trouver un document
user = collection.find_one({'username': 'john_doe'})
print(user)

# Trouver par ID
user = collection.find_one({'_id': ObjectId('507f1f77bcf86cd799439011')})

# Trouver tous les documents
users = collection.find()
for user in users:
    print(user)

# Trouver avec filtre
users = collection.find({'age': {'$gt': 25}})

# Projection (colonnes spécifiques)
users = collection.find({}, {'username': 1, 'email': 1, '_id': 0})

# Limit & Skip
users = collection.find().limit(10).skip(20)

# Sort
users = collection.find().sort('age', -1)  # -1 = DESC, 1 = ASC
users = collection.find().sort([('age', -1), ('username', 1)])

# Count
count = collection.count_documents({'age': {'$gt': 25}})
total = collection.count_documents({})

# === OPÉRATEURS DE REQUÊTE ===

# Comparaison
collection.find({'age': {'$eq': 30}})      # Égal
collection.find({'age': {'$ne': 30}})      # Différent
collection.find({'age': {'$gt': 25}})      # Plus grand
collection.find({'age': {'$gte': 25}})     # Plus grand ou égal
collection.find({'age': {'$lt': 40}})      # Plus petit
collection.find({'age': {'$lte': 40}})     # Plus petit ou égal
collection.find({'age': {'$in': [25, 30, 35]}})    # Dans liste
collection.find({'age': {'$nin': [25, 30]}})       # Pas dans liste

# Logique
collection.find({
    '$and': [
        {'age': {'$gt': 20}},
        {'age': {'$lt': 40}}
    ]
})

collection.find({
    '$or': [
        {'age': {'$lt': 20}},
        {'age': {'$gt': 60}}
    ]
})

collection.find({'age': {'$not': {'$gt': 30}}})

# Existence
collection.find({'email': {'$exists': True}})

# Type
collection.find({'age': {'$type': 'int'}})

# Regex
collection.find({'username': {'$regex': '^john'}})
collection.find({'username': {'$regex': 'doe, '$options': 'i'}})  # Case insensitive

# Array
collection.find({'tags': 'python'})                    # Contient
collection.find({'tags': {'$all': ['python', 'mongodb']}})  # Contient tous
collection.find({'tags': {'$size': 3}})                # Taille exacte

# Sous-documents
collection.find({'address.city': 'New York'})

# === MISE À JOUR ===

# Update un document
collection.update_one(
    {'username': 'john_doe'},
    {'$set': {'age': 31}}
)

# Update plusieurs documents
collection.update_many(
    {'age': {'$lt': 18}},
    {'$set': {'is_minor': True}}
)

# Opérateurs de mise à jour
collection.update_one(
    {'username': 'john_doe'},
    {
        '$set': {'email': 'newemail@example.com'},      # Définir valeur
        '$inc': {'age': 1},                              # Incrémenter
        '$mul': {'score': 2},                            # Multiplier
        '$rename': {'username': 'user_name'},            # Renommer champ
        '$unset': {'temp_field': ''},                    # Supprimer champ
        '$min': {'min_score': 10},                       # Minimum
        '$max': {'max_score': 100},                      # Maximum
        '$currentDate': {'last_modified': True}          # Date actuelle
    }
)

# Array updates
collection.update_one(
    {'username': 'john_doe'},
    {
        '$push': {'tags': 'new_tag'},                    # Ajouter à array
        '$pull': {'tags': 'old_tag'},                    # Retirer de array
        '$addToSet': {'tags': 'unique_tag'},             # Ajouter si unique
        '$pop': {'tags': 1}                              # Retirer dernier (1) ou premier (-1)
    }
)

# Update avec upsert
result = collection.update_one(
    {'username': 'new_user'},
    {'$set': {'email': 'new@example.com'}},
    upsert=True
)

# Replace document
collection.replace_one(
    {'username': 'john_doe'},
    {'username': 'john_doe', 'email': 'new@example.com', 'age': 31}
)

# === SUPPRESSION ===

# Supprimer un document
collection.delete_one({'username': 'john_doe'})

# Supprimer plusieurs documents
result = collection.delete_many({'age': {'$lt': 18}})
print(f"{result.deleted_count} documents supprimés")

# Supprimer tous les documents
collection.delete_many({})

# === AGRÉGATION ===

# Pipeline d'agrégation
pipeline = [
    {'$match': {'age': {'$gte': 25}}},
    {'$group': {
        '_id': '$city',
        'avg_age': {'$avg': '$age'},
        'count': {'$sum': 1}
    }},
    {'$sort': {'count': -1}},
    {'$limit': 10}
]
results = collection.aggregate(pipeline)

# Exemples d'opérateurs d'agrégation
pipeline = [
    # Match: filtrer
    {'$match': {'status': 'active'}},
    
    # Project: sélectionner champs
    {'$project': {
        'username': 1,
        'age': 1,
        'year': {'$year': '$created_at'}
    }},
    
    # Group: grouper
    {'$group': {
        '_id': '$category',
        'total': {'$sum': '$amount'},
        'avg': {'$avg': '$price'},
        'min': {'$min': '$price'},
        'max': {'$max': '$price'},
        'count': {'$sum': 1}
    }},
    
    # Sort: trier
    {'$sort': {'total': -1}},
    
    # Limit: limiter
    {'$limit': 10},
    
    # Skip: sauter
    {'$skip': 20},
    
    # Lookup: jointure
    {'$lookup': {
        'from': 'orders',
        'localField': '_id',
        'foreignField': 'user_id',
        'as': 'orders'
    }},
    
    # Unwind: décomposer array
    {'$unwind': '$tags'},
    
    # AddFields: ajouter champs
    {'$addFields': {
        'full_name': {'$concat': ['$first_name', ' ', '$last_name']}
    }}
]

# === INDEX ===

# Créer index simple
collection.create_index('username')

# Index composé
collection.create_index([('username', 1), ('email', 1)])

# Index unique
collection.create_index('email', unique=True)

# Index texte (recherche full-text)
collection.create_index([('description', 'text')])

# Recherche texte
collection.find({'$text': {'$search': 'python mongodb'}})

# Lister index
indexes = collection.list_indexes()

# Supprimer index
collection.drop_index('username_1')

# === TRANSACTIONS ===

# Transaction (MongoDB 4.0+)
with client.start_session() as session:
    with session.start_transaction():
        try:
            collection.insert_one({'username': 'test'}, session=session)
            collection.update_one(
                {'username': 'other'},
                {'$inc': {'balance': -100}},
                session=session
            )
            session.commit_transaction()
        except Exception as e:
            session.abort_transaction()
            print(f"Transaction annulée: {e}")

# === BULK OPERATIONS ===

from pymongo import InsertOne, UpdateOne, DeleteOne, ReplaceOne

requests = [
    InsertOne({'username': 'user1', 'email': 'user1@example.com'}),
    UpdateOne({'username': 'user2'}, {'$set': {'age': 30}}),
    DeleteOne({'username': 'user3'}),
    ReplaceOne({'username': 'user4'}, {'username': 'user4', 'email': 'new@example.com'})
]

result = collection.bulk_write(requests)
print(f"Insérés: {result.inserted_count}")
print(f"Modifiés: {result.modified_count}")
print(f"Supprimés: {result.deleted_count}")

# === EXEMPLE CLASSE ===

class MongoDB:
    def __init__(self, uri, database_name):
        self.client = MongoClient(uri)
        self.db = self.client[database_name]
    
    def insert(self, collection_name, document):
        collection = self.db[collection_name]
        result = collection.insert_one(document)
        return str(result.inserted_id)
    
    def find_one(self, collection_name, query):
        collection = self.db[collection_name]
        return collection.find_one(query)
    
    def find_all(self, collection_name, query=None, limit=0):
        collection = self.db[collection_name]
        cursor = collection.find(query or {})
        if limit > 0:
            cursor = cursor.limit(limit)
        return list(cursor)
    
    def update(self, collection_name, query, update_data):
        collection = self.db[collection_name]
        result = collection.update_one(query, {'$set': update_data})
        return result.modified_count
    
    def delete(self, collection_name, query):
        collection = self.db[collection_name]
        result = collection.delete_one(query)
        return result.deleted_count
    
    def close(self):
        self.client.close()


[OK] REDIS (REDIS-PY)

# === INSTALLATION ===
# pip install redis

import redis
from redis.exceptions import RedisError

# === CONNEXION ===

# Connexion simple
r = redis.Redis(host='localhost', port=6379, db=0)

# Connexion avec mot de passe
r = redis.Redis(
    host='localhost',
    port=6379,
    password='mypassword',
    db=0,
    decode_responses=True  # Décoder bytes en string
)

# Connection pool
pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
r = redis.Redis(connection_pool=pool)

# URL connection
r = redis.from_url('redis://localhost:6379/0')

# Vérifier connexion
try:
    r.ping()
    print("Connecté à Redis")
except RedisError:
    print("Échec de connexion")

# === STRINGS ===

# SET & GET
r.set('username', 'john_doe')
username = r.get('username')  # b'john_doe' ou 'john_doe' si decode_responses=True

# SET avec options
r.set('key', 'value', ex=60)      # Expire en 60 secondes
r.set('key', 'value', px=60000)   # Expire en 60000 millisecondes
r.set('key', 'value', nx=True)    # Set seulement si n'existe pas
r.set('key', 'value', xx=True)    # Set seulement si existe

# SETEX (SET avec expiration)
r.setex('session', 3600, 'session_data')

# SETNX (SET if Not eXists)
r.setnx('lock', 'locked')

# MSET & MGET (multiple)
r.mset({'key1': 'value1', 'key2': 'value2'})
values = r.mget('key1', 'key2')

# INCR & DECR
r.set('counter', 0)
r.incr('counter')        # 1
r.incrby('counter', 5)   # 6
r.decr('counter')        # 5
r.decrby('counter', 2)   # 3

# APPEND
r.append('message', 'Hello ')
r.append('message', 'World')

# STRLEN
length = r.strlen('message')

# === LISTES ===

# LPUSH & RPUSH (ajouter)
r.lpush('tasks', 'task1')      # Ajouter à gauche
r.rpush('tasks', 'task2')      # Ajouter à droite
r.rpush('tasks', 'task3', 'task4', 'task5')

# LPOP & RPOP (retirer)
task = r.lpop('tasks')         # Retirer de gauche
task = r.rpop('tasks')         # Retirer de droite

# LRANGE (obtenir plage)
tasks = r.lrange('tasks', 0, -1)  # Tous les éléments
tasks = r.lrange('tasks', 0, 4)   # 5 premiers

# LLEN (longueur)
length = r.llen('tasks')

# LINDEX (élément par index)
task = r.lindex('tasks', 0)

# LSET (définir élément)
r.lset('tasks', 0, 'updated_task')

# LREM (supprimer)
r.lrem('tasks', 1, 'task2')    # Supprimer 1 occurrence

# LTRIM (garder plage)
r.ltrim('tasks', 0, 9)         # Garder 10 premiers

# BLPOP & BRPOP (blocking pop)
task = r.blpop('tasks', timeout=5)  # Attendre max 5 secondes

# === SETS ===

# SADD (ajouter)
r.sadd('tags', 'python', 'mongodb', 'redis')

# SMEMBERS (tous les membres)
tags = r.smembers('tags')

# SISMEMBER (vérifier membre)
exists = r.sismember('tags', 'python')

# SCARD (nombre de membres)
count = r.scard('tags')

# SREM (supprimer)
r.srem('tags', 'mongodb')

# SPOP (retirer aléatoire)
tag = r.spop('tags')

# SRANDMEMBER (membre aléatoire sans retirer)
tag = r.srandmember('tags')

# Opérations d'ensemble
r.sadd('set1', 'a', 'b', 'c')
r.sadd('set2', 'b', 'c', 'd')

union = r.sunion('set1', 'set2')           # Union
inter = r.sinter('set1', 'set2')           # Intersection
diff = r.sdiff('set1', 'set2')             # Différence

# === SORTED SETS (ZSET) ===

# ZADD (ajouter avec score)
r.zadd('leaderboard', {'player1': 100, 'player2': 200, 'player3': 150})

# ZRANGE (plage par rang)
players = r.zrange('leaderboard', 0, -1)            # Ordre croissant
players = r.zrange('leaderboard', 0, -1, withscores=True)

# ZREVRANGE (plage ordre décroissant)
players = r.zrevrange('leaderboard', 0, 9)          # Top 10

# ZRANGEBYSCORE (plage par score)
players = r.zrangebyscore('leaderboard', 100, 200)

# ZSCORE (obtenir score)
score = r.zscore('leaderboard', 'player1')

# ZINCRBY (incrémenter score)
r.zincrby('leaderboard', 50, 'player1')

# ZRANK & ZREVRANK (rang)
rank = r.zrank('leaderboard', 'player1')           # 0-based
rank = r.zrevrank('leaderboard', 'player1')        # Ordre inverse

# ZCARD (nombre d'éléments)
count = r.zcard('leaderboard')

# ZREM (supprimer)
r.zrem('leaderboard', 'player1')

# === HASHES ===

# HSET & HGET
r.hset('user:1', 'username', 'john_doe')
r.hset('user:1', 'email', 'john@example.com')
username = r.hget('user:1', 'username')

# HSET multiple
r.hset('user:1', mapping={
    'username': 'john_doe',
    'email': 'john@example.com',
    'age': 30
})

# HMSET & HMGET (deprecated, use hset)
r.hset('user:2', mapping={'username': 'jane', 'email': 'jane@example.com'})
data = r.hmget('user:2', 'username', 'email')

# HGETALL (tous les champs)
user = r.hgetall('user:1')

# HKEYS & HVALS
keys = r.hkeys('user:1')
values = r.hvals('user:1')

# HEXISTS (vérifier champ)
exists = r.hexists('user:1', 'username')

# HDEL (supprimer champ)
r.hdel('user:1', 'age')

# HINCRBY (incrémenter)
r.hincrby('user:1', 'login_count', 1)

# HLEN (nombre de champs)
count = r.hlen('user:1')

# === CLÉS ===

# EXISTS
exists = r.exists('key')
exists = r.exists('key1', 'key2', 'key3')  # Nombre de clés existantes

# DEL
r.delete('key')
r.delete('key1', 'key2', 'key3')

# EXPIRE & TTL
r.expire('key', 60)          # Expire dans 60 secondes
ttl = r.ttl('key')           # Temps restant (-1 = pas d'expiration, -2 = n'existe pas)

# PERSIST (supprimer expiration)
r.persist('key')

# KEYS (lister clés par pattern)
keys = r.keys('user:*')
keys = r.keys('*')           # ATTENTION: lent sur grande base

# SCAN (itérer sur clés - recommandé)
cursor = 0
while True:
    cursor, keys = r.scan(cursor, match='user:*', count=100)
    for key in keys:
        print(key)
    if cursor == 0:
        break

# TYPE
key_type = r.type('key')

# RENAME
r.rename('old_key', 'new_key')

# === TRANSACTIONS ===

# Pipeline (grouper commandes)
pipe = r.pipeline()
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.incr('counter')
results = pipe.execute()

# Transaction (MULTI/EXEC)
pipe = r.pipeline(transaction=True)
pipe.multi()
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.execute()

# Watch (optimistic locking)
with r.pipeline() as pipe:
    while True:
        try:
            pipe.watch('balance')
            current = int(pipe.get('balance'))
            pipe.multi()
            pipe.set('balance', current + 100)
            pipe.execute()
            break
        except redis.WatchError:
            continue

# === PUB/SUB ===

# Publisher
r.publish('news', 'Breaking news!')

# Subscriber
pubsub = r.pubsub()
pubsub.subscribe('news', 'updates')

for message in pubsub.listen():
    if message['type'] == 'message':
        print(message['data'])

# Pattern subscribe
pubsub.psubscribe('news:*')

# Unsubscribe
pubsub.unsubscribe('news')

# === EXEMPLE CLASSE ===

class RedisCache:
    def __init__(self, host='localhost', port=6379, db=0):
        self.r = redis.Redis(
            host=host,
            port=port,
            db=db,
            decode_responses=True
        )
    
    def set(self, key, value, expire=None):
        return self.r.set(key, value, ex=expire)
    
    def get(self, key):
        return self.r.get(key)
    
    def delete(self, key):
        return self.r.delete(key)
    
    def exists(self, key):
        return self.r.exists(key)
    
    def incr(self, key, amount=1):
        return self.r.incrby(key, amount)
    
    def set_hash(self, name, key, value):
        return self.r.hset(name, key, value)
    
    def get_hash(self, name, key=None):
        if key:
            return self.r.hget(name, key)
        return self.r.hgetall(name)
    
    def add_to_list(self, name, *values):
        return self.r.rpush(name, *values)
    
    def get_list(self, name, start=0, end=-1):
        return self.r.lrange(name, start, end)
    
    def close(self):
        self.r.close()


[OK] CONNEXION POOLS & GESTION

# === SQLITE3 POOL ===

from contextlib import contextmanager
import sqlite3
import threading

class SQLitePool:
    def __init__(self, database, size=5):
        self.database = database
        self.size = size
        self.pool = []
        self.lock = threading.Lock()
        self._init_pool()
    
    def _init_pool(self):
        for _ in range(self.size):
            conn = sqlite3.connect(self.database, check_same_thread=False)
            self.pool.append(conn)
    
    @contextmanager
    def get_connection(self):
        with self.lock:
            if not self.pool:
                conn = sqlite3.connect(self.database, check_same_thread=False)
            else:
                conn = self.pool.pop()
        try:
            yield conn
        finally:
            with self.lock:
                self.pool.append(conn)

# === SQLALCHEMY POOL ===

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool, NullPool, StaticPool

# Pool avec queue
engine = create_engine(
    'postgresql://user:pass@localhost/db',
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=3600,
    pool_pre_ping=True
)

# Sans pool (connexion par requête)
engine = create_engine(
    'postgresql://user:pass@localhost/db',
    poolclass=NullPool
)

# === CONNECTION CONTEXT MANAGERS ===

@contextmanager
def get_db_connection():
    conn = sqlite3.connect('database.db')
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Utilisation
with get_db_connection() as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users')


[OK] MIGRATIONS DE BASE DE DONNÉES

# === ALEMBIC (pour SQLAlchemy) ===

# Installation
# pip install alembic

# Initialiser
# alembic init alembic

# alembic.ini - Configuration
sqlalchemy.url = postgresql://user:pass@localhost/db

# alembic/env.py - Importer modèles
from myapp.models import Base
target_metadata = Base.metadata

# Créer migration
# alembic revision --autogenerate -m "create users table"

# Appliquer migrations
# alembic upgrade head

# Revenir en arrière
# alembic downgrade -1

# Voir historique
# alembic history

# === DJANGO ORM MIGRATIONS ===

# Créer migrations
# python manage.py makemigrations

# Appliquer migrations
# python manage.py migrate

# Voir migrations
# python manage.py showmigrations

# Revenir en arrière
# python manage.py migrate app_name 0003

# === MIGRATION MANUELLE SQLite ===

def migrate_database():
    conn = sqlite3.connect('database.db')
    cursor = conn.cursor()
    
    # Version 1: Créer table users
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS schema_version (
            version INTEGER PRIMARY KEY
        )
    ''')
    
    cursor.execute('SELECT version FROM schema_version')
    row = cursor.fetchone()
    current_version = row[0] if row else 0
    
    if current_version < 1:
        cursor.execute('''
            CREATE TABLE users (
                id INTEGER PRIMARY KEY,
                username TEXT NOT NULL
            )
        ''')
        cursor.execute('INSERT INTO schema_version VALUES (1)')
    
    if current_version < 2:
        cursor.execute('ALTER TABLE users ADD COLUMN email TEXT')
        cursor.execute('UPDATE schema_version SET version = 2')
    
    conn.commit()
    conn.close()


[OK] BONNES PRATIQUES

# 1. Utiliser context managers
with sqlite3.connect('database.db') as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users')

# 2. Paramètres préparés (éviter SQL injection)
# [OK] BON
cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
cursor.execute('SELECT * FROM users WHERE name = %s', (name,))

# [X] MAUVAIS
cursor.execute(f'SELECT * FROM users WHERE id = {user_id}')

# 3. Gestion d'erreurs
try:
    conn = sqlite3.connect('database.db')
    cursor = conn.cursor()
    cursor.execute('INSERT INTO users VALUES (?, ?)', (name, email))
    conn.commit()
except sqlite3.IntegrityError as e:
    print(f"Erreur d'intégrité: {e}")
except sqlite3.Error as e:
    print(f"Erreur SQL: {e}")
    conn.rollback()
finally:
    conn.close()

# 4. Index pour performance
cursor.execute('CREATE INDEX idx_username ON users(username)')

# 5. Transactions pour cohérence
conn.execute('BEGIN')
try:
    cursor.execute('INSERT INTO users ...')
    cursor.execute('INSERT INTO profiles ...')
    conn.commit()
except:
    conn.rollback()

# 6. Connection pooling pour apps web
engine = create_engine('postgresql://...', pool_size=10)

# 7. Lazy loading vs Eager loading
# Lazy (charge au besoin)
user = session.query(User).get(1)
print(user.orders)  # Query ici

# Eager (charge d'avance)
user = session.query(User).options(joinedload(User.orders)).get(1)

# 8. Fermer connexions
conn.close()  # Toujours fermer!

# 9. Utiliser ORM pour apps complexes
# SQLAlchemy pour abstraction et portabilité

# 10. Sauvegardes régulières
def backup_sqlite(source, dest):
    src = sqlite3.connect(source)
    dst = sqlite3.connect(dest)
    src.backup(dst)
    dst.close()
    src.close()


[OK] PERFORMANCE & OPTIMISATION

# === SQLITE ===

# WAL mode (Write-Ahead Logging)
conn.execute('PRAGMA journal_mode=WAL')

# Optimiser performance
conn.execute('PRAGMA synchronous = NORMAL')
conn.execute('PRAGMA cache_size = 10000')
conn.execute('PRAGMA temp_store = MEMORY')

# Analyser requêtes
cursor.execute('EXPLAIN QUERY PLAN SELECT * FROM users WHERE age > 25')

# Vacuum (compacter)
conn.execute('VACUUM')

# === POSTGRESQL ===

# Analyser requêtes
cursor.execute('EXPLAIN ANALYZE SELECT * FROM users WHERE age > 25')

# Créer index pour colonnes fréquentes
cursor.execute('CREATE INDEX CONCURRENTLY idx_age ON users(age)')

# Vacuum (nettoyer)
conn.autocommit = True
cursor.execute('VACUUM ANALYZE users')

# === MYSQL ===

# Analyser requêtes
cursor.execute('EXPLAIN SELECT * FROM users WHERE age > 25')

# Optimiser table
cursor.execute('OPTIMIZE TABLE users')

# Analyser table
cursor.execute('ANALYZE TABLE users')

# === MONGODB ===

# Créer index pour requêtes fréquentes
collection.create_index([('username', 1), ('email', 1)])

# Analyser performance requête
cursor = collection.find({'age': {'$gt': 25}})
explain = cursor.explain()

# === SQLALCHEMY ===

# Eager loading pour éviter N+1 queries
from sqlalchemy.orm import joinedload, subqueryload

users = session.query(User).options(
    joinedload(User.orders)
).all()

# Batch insert
session.bulk_insert_mappings(User, [
    {'username': 'user1', 'email': 'user1@example.com'},
    {'username': 'user2', 'email': 'user2@example.com'},
])

# Query profiling
from sqlalchemy import event
from sqlalchemy.engine import Engine
import time

@event.listens_for(Engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, params, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time.time())

@event.listens_for(Engine, "after_cursor_execute")
def receive_after_cursor_execute(conn, cursor, statement, params, context, executemany):
    total = time.time() - conn.info['query_start_time'].pop(-1)
    print(f"Query: {statement[:50]}... - Time: {total:.4f}s")


[OK] ASYNC/AWAIT (BASES DE DONNÉES ASYNCHRONES)

# === AIOSQLITE (SQLite async) ===

# Installation
# pip install aiosqlite

import asyncio
import aiosqlite

async def create_users_table():
    async with aiosqlite.connect('database.db') as db:
        await db.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY,
                username TEXT NOT NULL,
                email TEXT NOT NULL
            )
        ''')
        await db.commit()

async def insert_user(username, email):
    async with aiosqlite.connect('database.db') as db:
        await db.execute(
            'INSERT INTO users (username, email) VALUES (?, ?)',
            (username, email)
        )
        await db.commit()

async def get_all_users():
    async with aiosqlite.connect('database.db') as db:
        async with db.execute('SELECT * FROM users') as cursor:
            rows = await cursor.fetchall()
            return rows

# Exécution
async def main():
    await create_users_table()
    await insert_user('john_doe', 'john@example.com')
    users = await get_all_users()
    print(users)

asyncio.run(main())

# === AIOMYSQL (MySQL async) ===

# Installation
# pip install aiomysql

import aiomysql

async def mysql_example():
    conn = await aiomysql.connect(
        host='localhost',
        port=3306,
        user='root',
        password='password',
        db='mydb'
    )
    
    async with conn.cursor() as cursor:
        await cursor.execute('SELECT * FROM users')
        result = await cursor.fetchall()
        print(result)
    
    conn.close()

# Pool de connexions
async def mysql_pool_example():
    pool = await aiomysql.create_pool(
        host='localhost',
        port=3306,
        user='root',
        password='password',
        db='mydb',
        minsize=1,
        maxsize=10
    )
    
    async with pool.acquire() as conn:
        async with conn.cursor() as cursor:
            await cursor.execute('SELECT * FROM users')
            result = await cursor.fetchall()
            print(result)
    
    pool.close()
    await pool.wait_closed()

# === ASYNCPG (PostgreSQL async) ===

# Installation
# pip install asyncpg

import asyncpg

async def postgres_example():
    conn = await asyncpg.connect(
        host='localhost',
        port=5432,
        user='postgres',
        password='password',
        database='mydb'
    )
    
    # Requête simple
    rows = await conn.fetch('SELECT * FROM users')
    for row in rows:
        print(row['username'], row['email'])
    
    # Insert
    await conn.execute(
        'INSERT INTO users (username, email) VALUES ($1, $2)',
        'john_doe', 'john@example.com'
    )
    
    # Fetchone
    row = await conn.fetchrow('SELECT * FROM users WHERE id = $1', 1)
    
    await conn.close()

# Pool de connexions
async def postgres_pool_example():
    pool = await asyncpg.create_pool(
        host='localhost',
        user='postgres',
        password='password',
        database='mydb',
        min_size=10,
        max_size=20
    )
    
    async with pool.acquire() as conn:
        rows = await conn.fetch('SELECT * FROM users')
        print(rows)
    
    await pool.close()

# Transaction
async def postgres_transaction():
    conn = await asyncpg.connect('postgresql://user:pass@localhost/db')
    
    async with conn.transaction():
        await conn.execute('INSERT INTO users (username) VALUES ($1)', 'user1')
        await conn.execute('INSERT INTO users (username) VALUES ($1)', 'user2')
    
    await conn.close()

# === MOTOR (MongoDB async) ===

# Installation
# pip install motor

import motor.motor_asyncio

async def mongodb_example():
    client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017')
    db = client['mydb']
    collection = db['users']
    
    # Insert
    result = await collection.insert_one({
        'username': 'john_doe',
        'email': 'john@example.com'
    })
    print(f"Inserted ID: {result.inserted_id}")
    
    # Find
    users = []
    async for user in collection.find({'age': {'$gt': 25}}):
        users.append(user)
    
    # Find one
    user = await collection.find_one({'username': 'john_doe'})
    
    # Update
    await collection.update_one(
        {'username': 'john_doe'},
        {'$set': {'age': 31}}
    )
    
    # Delete
    await collection.delete_one({'username': 'john_doe'})
    
    client.close()

# === AIOREDIS (Redis async) ===

# Installation
# pip install aioredis

import aioredis

async def redis_example():
    redis = await aioredis.create_redis_pool('redis://localhost')
    
    # Set & Get
    await redis.set('key', 'value')
    value = await redis.get('key', encoding='utf-8')
    
    # List operations
    await redis.rpush('mylist', 'item1', 'item2')
    items = await redis.lrange('mylist', 0, -1)
    
    # Hash operations
    await redis.hset('user:1', 'username', 'john_doe')
    username = await redis.hget('user:1', 'username', encoding='utf-8')
    
    redis.close()
    await redis.wait_closed()

# === SQLALCHEMY ASYNC ===

# Installation
# pip install sqlalchemy[asyncio] aiosqlite

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, select

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    email = Column(String(100))

# Créer engine async
engine = create_async_engine('sqlite+aiosqlite:///database.db', echo=True)

# Session factory
async_session = async_sessionmaker(engine, class_=AsyncSession)

# Créer tables
async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# Insérer
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

# Sélectionner
async def get_all_users():
    async with async_session() as session:
        result = await session.execute(select(User))
        users = result.scalars().all()
        return users

# Mise à jour
async def update_user(user_id, new_email):
    async with async_session() as session:
        result = await session.execute(select(User).where(User.id == user_id))
        user = result.scalar_one()
        user.email = new_email
        await session.commit()

# Exemple complet
async def main():
    await init_db()
    await create_user('john_doe', 'john@example.com')
    users = await get_all_users()
    for user in users:
        print(user.username, user.email)

asyncio.run(main())


[OK] ORM ALTERNATIFS

# === PEEWEE (ORM Simple) ===

# Installation
# pip install peewee

from peewee import *

db = SqliteDatabase('database.db')

class BaseModel(Model):
    class Meta:
        database = db

class User(BaseModel):
    username = CharField(unique=True)
    email = CharField()
    age = IntegerField(null=True)
    is_active = BooleanField(default=True)

class Order(BaseModel):
    user = ForeignKeyField(User, backref='orders')
    product = CharField()
    quantity = IntegerField()

# Créer tables
db.connect()
db.create_tables([User, Order])

# Insérer
user = User.create(username='john_doe', email='john@example.com', age=30)

# Sélectionner
users = User.select().where(User.age > 25)
for user in users:
    print(user.username)

# Get or create
user, created = User.get_or_create(
    username='jane_doe',
    defaults={'email': 'jane@example.com'}
)

# Mise à jour
query = User.update(age=31).where(User.username == 'john_doe')
query.execute()

# Suppression
User.delete().where(User.age < 18).execute()

# Relations
user = User.get(User.username == 'john_doe')
for order in user.orders:
    print(order.product)

# Jointures
query = (User
         .select(User, Order)
         .join(Order)
         .where(User.age > 25))

# Agrégation
from peewee import fn
result = (User
          .select(fn.COUNT(User.id).alias('count'))
          .scalar())

# === TORTOISE ORM (Async) ===

# Installation
# pip install tortoise-orm

from tortoise import Tortoise, fields
from tortoise.models import Model

class User(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=50, unique=True)
    email = fields.CharField(max_length=100)
    age = fields.IntField(null=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    
    orders: fields.ReverseRelation["Order"]
    
    class Meta:
        table = "users"

class Order(Model):
    id = fields.IntField(pk=True)
    user = fields.ForeignKeyField('models.User', related_name='orders')
    product = fields.CharField(max_length=100)
    quantity = fields.IntField()
    
    class Meta:
        table = "orders"

# Initialiser
async def init():
    await Tortoise.init(
        db_url='sqlite://database.db',
        modules={'models': ['__main__']}
    )
    await Tortoise.generate_schemas()

# Créer
async def create_user():
    user = await User.create(username='john_doe', email='john@example.com')
    return user

# Sélectionner
async def get_users():
    users = await User.filter(age__gt=25)
    return users

# Get or create
user, created = await User.get_or_create(
    username='jane_doe',
    defaults={'email': 'jane@example.com'}
)

# Mise à jour
await User.filter(username='john_doe').update(age=31)

# Suppression
await User.filter(age__lt=18).delete()

# Relations
user = await User.get(username='john_doe').prefetch_related('orders')
for order in user.orders:
    print(order.product)

# === PONY ORM (Simple & Pythonique) ===

# Installation
# pip install pony

from pony.orm import *

db = Database()

class User(db.Entity):
    username = Required(str, unique=True)
    email = Required(str)
    age = Optional(int)
    orders = Set('Order')

class Order(db.Entity):
    user = Required(User)
    product = Required(str)
    quantity = Required(int)

db.bind('sqlite', 'database.db', create_db=True)
db.generate_mapping(create_tables=True)

# Utiliser avec décorateur @db_session
@db_session
def create_user(username, email):
    return User(username=username, email=email)

@db_session
def get_users():
    # Syntaxe Python native!
    users = select(u for u in User if u.age > 25)[:]
    return users

# Generator expressions
@db_session
def query_example():
    # Très pythonique
    users = User.select(lambda u: u.age > 25 and u.username.startswith('j'))
    
    # Avec order_by
    users = select(u for u in User if u.age > 25).order_by(User.age)
    
    # Count
    count = count(u for u in User if u.age > 25)

# Relations
@db_session
def get_user_orders():
    user = User.get(username='john_doe')
    for order in user.orders:
        print(order.product)


[OK] QUERY BUILDERS

# === PYPIKA (Query Builder) ===

# Installation
# pip install pypika

from pypika import Query, Table, Field
from pypika import functions as fn

users = Table('users')

# SELECT
query = Query.from_(users).select('*')
# SELECT * FROM users

query = Query.from_(users).select(users.username, users.email)
# SELECT username, email FROM users

# WHERE
query = Query.from_(users).select('*').where(users.age > 25)
# SELECT * FROM users WHERE age > 25

# JOIN
orders = Table('orders')
query = (Query
         .from_(users)
         .join(orders).on(users.id == orders.user_id)
         .select(users.username, orders.product))

# Agrégation
query = (Query
         .from_(users)
         .select(users.city, fn.Count('*').as_('count'))
         .groupby(users.city))

# === SQL-QUERIES (Chaîne SQL) ===

# Installation
# pip install sql-queries

from sql_queries import *

# Construire requête
query = (
    Select('username', 'email')
    .from_('users')
    .where(Field('age') > 25)
    .order_by('username')
    .limit(10)
)

sql_string = str(query)


[OK] DATA VALIDATION & SERIALIZATION

# === PYDANTIC (avec SQLAlchemy) ===

# Installation
# pip install pydantic

from pydantic import BaseModel, EmailStr, validator
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    username: str
    email: EmailStr
    age: Optional[int] = None
    
    @validator('age')
    def age_must_be_positive(cls, v):
        if v is not None and v < 0:
            raise ValueError('Age must be positive')
        return v

class UserCreate(UserBase):
    password: str

class UserResponse(UserBase):
    id: int
    created_at: datetime
    
    class Config:
        orm_mode = True  # Permet conversion depuis ORM

# Utilisation avec SQLAlchemy
def create_user(user_data: UserCreate):
    user = User(**user_data.dict())
    session.add(user)
    session.commit()
    return UserResponse.from_orm(user)

# === MARSHMALLOW (Serialization) ===

# Installation
# pip install marshmallow

from marshmallow import Schema, fields, validate, post_load

class UserSchema(Schema):
    id = fields.Int(dump_only=True)
    username = fields.Str(required=True, validate=validate.Length(min=3))
    email = fields.Email(required=True)
    age = fields.Int(validate=validate.Range(min=0, max=150))
    created_at = fields.DateTime(dump_only=True)
    
    @post_load
    def make_user(self, data, **kwargs):
        return User(**data)

# Serialization (object -> dict)
user = User(id=1, username='john', email='john@example.com')
schema = UserSchema()
result = schema.dump(user)

# Deserialization (dict -> object)
data = {'username': 'john', 'email': 'john@example.com', 'age': 30}
user = schema.load(data)


[OK] BACKUP & RESTORE

# === SQLITE BACKUP ===

import sqlite3
import shutil
from datetime import datetime

def backup_sqlite(source_db, backup_dir):
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_file = f"{backup_dir}/backup_{timestamp}.db"
    
    # Méthode 1: Copie fichier
    shutil.copy2(source_db, backup_file)
    
    # Méthode 2: API backup
    source = sqlite3.connect(source_db)
    backup = sqlite3.connect(backup_file)
    source.backup(backup)
    backup.close()
    source.close()

def restore_sqlite(backup_file, target_db):
    shutil.copy2(backup_file, target_db)

# === MYSQL BACKUP ===

import subprocess

def backup_mysql(host, user, password, database, backup_file):
    cmd = [
        'mysqldump',
        f'--host={host}',
        f'--user={user}',
        f'--password={password}',
        database
    ]
    
    with open(backup_file, 'w') as f:
        subprocess.run(cmd, stdout=f)

def restore_mysql(host, user, password, database, backup_file):
    cmd = [
        'mysql',
        f'--host={host}',
        f'--user={user}',
        f'--password={password}',
        database
    ]
    
    with open(backup_file, 'r') as f:
        subprocess.run(cmd, stdin=f)

# === POSTGRESQL BACKUP ===

def backup_postgres(host, user, database, backup_file):
    cmd = [
        'pg_dump',
        f'--host={host}',
        f'--username={user}',
        f'--file={backup_file}',
        database
    ]
    subprocess.run(cmd)

def restore_postgres(host, user, database, backup_file):
    cmd = [
        'psql',
        f'--host={host}',
        f'--username={user}',
        f'--dbname={database}',
        f'--file={backup_file}'
    ]
    subprocess.run(cmd)

# === MONGODB BACKUP ===

def backup_mongodb(host, database, backup_dir):
    cmd = [
        'mongodump',
        f'--host={host}',
        f'--db={database}',
        f'--out={backup_dir}'
    ]
    subprocess.run(cmd)

def restore_mongodb(host, database, backup_dir):
    cmd = [
        'mongorestore',
        f'--host={host}',
        f'--db={database}',
        backup_dir
    ]
    subprocess.run(cmd)


[OK] TESTING

# === PYTEST AVEC BASES DE DONNÉES ===

# Installation
# pip install pytest pytest-asyncio

import pytest
import sqlite3

@pytest.fixture
def db_connection():
    conn = sqlite3.connect(':memory:')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE users (
            id INTEGER PRIMARY KEY,
            username TEXT NOT NULL,
            email TEXT NOT NULL
        )
    ''')
    conn.commit()
    yield conn
    conn.close()

def test_insert_user(db_connection):
    cursor = db_connection.cursor()
    cursor.execute(
        'INSERT INTO users (username, email) VALUES (?, ?)',
        ('john_doe', 'john@example.com')
    )
    db_connection.commit()
    
    cursor.execute('SELECT * FROM users WHERE username = ?', ('john_doe',))
    user = cursor.fetchone()
    assert user is not None
    assert user[1] == 'john_doe'

# === MOCK DATABASE ===

from unittest.mock import Mock, patch

def test_with_mock():
    mock_cursor = Mock()
    mock_cursor.fetchall.return_value = [
        (1, 'john_doe', 'john@example.com'),
        (2, 'jane_doe', 'jane@example.com')
    ]
    
    mock_conn = Mock()
    mock_conn.cursor.return_value = mock_cursor
    
    with patch('sqlite3.connect', return_value=mock_conn):
        # Test code here
        pass

# === FACTORY BOY (Test Data) ===

# Installation
# pip install factory_boy

import factory
from factory.alchemy import SQLAlchemyModelFactory

class UserFactory(SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session = session
    
    username = factory.Sequence(lambda n: f'user{n}')
    email = factory.LazyAttribute(lambda obj: f'{obj.username}@example.com')
    age = factory.Faker('random_int', min=18, max=80)

# Utilisation
user = UserFactory()
users = UserFactory.create_batch(10)

# === FAKER (Données de test) ===

# Installation
# pip install faker

from faker import Faker

fake = Faker()

def generate_test_users(count=10):
    users = []
    for _ in range(count):
        users.append({
            'username': fake.user_name(),
            'email': fake.email(),
            'age': fake.random_int(min=18, max=80),
            'address': fake.address(),
            'phone': fake.phone_number()
        })
    return users


[OK] SÉCURITÉ

# === PRÉVENTION SQL INJECTION ===

# [X] DANGEREUX
user_input = "admin' OR '1'='1"
cursor.execute(f"SELECT * FROM users WHERE username = '{user_input}'")

# [OK] SÉCURISÉ
cursor.execute("SELECT * FROM users WHERE username = ?", (user_input,))
cursor.execute("SELECT * FROM users WHERE username = %s", (user_input,))

# === HASHAGE MOTS DE PASSE ===

# Installation
# pip install bcrypt

import bcrypt

def hash_password(password: str) -> str:
    salt = bcrypt.gensalt()
    hashed = bcrypt.hashpw(password.encode('utf-8'), salt)
    return hashed.decode('utf-8')

def verify_password(password: str, hashed: str) -> bool:
    return bcrypt.checkpw(
        password.encode('utf-8'),
        hashed.encode('utf-8')
    )

# Utilisation
hashed = hash_password('my_password')
is_valid = verify_password('my_password', hashed)

# === CHIFFREMENT DONNÉES SENSIBLES ===

# Installation
# pip install cryptography

from cryptography.fernet import Fernet

# Générer clé
key = Fernet.generate_key()
cipher = Fernet(key)

# Chiffrer
encrypted = cipher.encrypt(b'sensitive data')

# Déchiffrer
decrypted = cipher.decrypt(encrypted)

# Stocker données chiffrées
def store_encrypted_data(data: str):
    encrypted = cipher.encrypt(data.encode())
    cursor.execute(
        'INSERT INTO secrets (data) VALUES (?)',
        (encrypted,)
    )

# === VARIABLES D'ENVIRONNEMENT ===

# Installation
# pip install python-dotenv

from dotenv import load_dotenv
import os

load_dotenv()

DB_HOST = os.getenv('DB_HOST')
DB_USER = os.getenv('DB_USER')
DB_PASSWORD = os.getenv('DB_PASSWORD')

# .env file
# DB_HOST=localhost
# DB_USER=myuser
# DB_PASSWORD=mypassword


[OK] MONITORING & LOGGING

# === LOGGING ===

import logging

# Configuration logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('database.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

# Utilisation
def create_user(username, email):
    try:
        cursor.execute(
            'INSERT INTO users (username, email) VALUES (?, ?)',
            (username, email)
        )
        conn.commit()
        logger.info(f"User created: {username}")
    except sqlite3.IntegrityError as e:
        logger.error(f"Failed to create user {username}: {e}")
        conn.rollback()

# === SQLALCHEMY LOGGING ===

import logging

logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# === MONITORING PERFORMANCE ===

import time
from functools import wraps

def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        logger.info(f"{func.__name__} took {end - start:.4f} seconds")
        return result
    return wrapper

@timer
def slow_query():
    cursor.execute('SELECT * FROM large_table')
    return cursor.fetchall()


[OK] EXEMPLES COMPLETS D'APPLICATIONS

# === APPLICATION CRUD COMPLÈTE (SQLite) ===

import sqlite3
from contextlib import contextmanager
from typing import List, Optional, Dict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Database:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.init_database()
    
    @contextmanager
    def get_connection(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception as e:
            conn.rollback()
            logger.error(f"Database error: {e}")
            raise
        finally:
            conn.close()
    
    def init_database(self):
        with self.get_connection() as conn:
            conn.execute('''
                CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    username TEXT NOT NULL UNIQUE,
                    email TEXT NOT NULL,
                    age INTEGER,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            ''')
            conn.execute('''
                CREATE INDEX IF NOT EXISTS idx_username 
                ON users(username)
            ''')
    
    def create_user(self, username: str, email: str, age: Optional[int] = None) -> int:
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
                (username, email, age)
            )
            logger.info(f"Created user: {username}")
            return cursor.lastrowid
    
    def get_user(self, user_id: int) -> Optional[Dict]:
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
            row = cursor.fetchone()
            return dict(row) if row else None
    
    def get_all_users(self, limit: int = 100, offset: int = 0) -> List[Dict]:
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                'SELECT * FROM users ORDER BY created_at DESC LIMIT ? OFFSET ?',
                (limit, offset)
            )
            return [dict(row) for row in cursor.fetchall()]
    
    def update_user(self, user_id: int, **kwargs) -> bool:
        if not kwargs:
            return False
        
        fields = ', '.join(f'{k} = ?' for k in kwargs.keys())
        values = list(kwargs.values()) + [user_id]
        
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                f'UPDATE users SET {fields} WHERE id = ?',
                values
            )
            logger.info(f"Updated user {user_id}")
            return cursor.rowcount > 0
    
    def delete_user(self, user_id: int) -> bool:
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
            logger.info(f"Deleted user {user_id}")
            return cursor.rowcount > 0
    
    def search_users(self, query: str) -> List[Dict]:
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                'SELECT * FROM users WHERE username LIKE ? OR email LIKE ?',
                (f'%{query}%', f'%{query}%')
            )
            return [dict(row) for row in cursor.fetchall()]

# Utilisation
if __name__ == '__main__':
    db = Database('app.db')
    
    # Créer utilisateurs
    user_id = db.create_user('john_doe', 'john@example.com', 30)
    print(f"Created user with ID: {user_id}")
    
    # Récupérer utilisateur
    user = db.get_user(user_id)
    print(f"User: {user}")
    
    # Mettre à jour
    db.update_user(user_id, age=31, email='newemail@example.com')
    
    # Rechercher
    results = db.search_users('john')
    print(f"Search results: {results}")
    
    # Lister tous
    users = db.get_all_users()
    for user in users:
        print(user)


# === API REST AVEC FASTAPI + SQLAlchemy ===

# Installation
# pip install fastapi uvicorn sqlalchemy

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel, EmailStr
from typing import List, Optional

# Configuration base de données
SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Modèle SQLAlchemy
class UserDB(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    age = Column(Integer, nullable=True)

Base.metadata.create_all(bind=engine)

# Schémas Pydantic
class UserBase(BaseModel):
    username: str
    email: EmailStr
    age: Optional[int] = None

class UserCreate(UserBase):
    pass

class UserUpdate(BaseModel):
    username: Optional[str] = None
    email: Optional[EmailStr] = None
    age: Optional[int] = None

class User(UserBase):
    id: int
    
    class Config:
        orm_mode = True

# FastAPI app
app = FastAPI(title="User API")

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Endpoints
@app.post("/users/", response_model=User, status_code=201)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(UserDB).filter(UserDB.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    new_user = UserDB(**user.dict())
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

@app.get("/users/", response_model=List[User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = db.query(UserDB).offset(skip).limit(limit).all()
    return users

@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    db_user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    update_data = user.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_user, key, value)
    
    db.commit()
    db.refresh(db_user)
    return db_user

@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(UserDB).filter(UserDB.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    db.delete(db_user)
    db.commit()
    return None

# Lancer: uvicorn main:app --reload


# === APPLICATION FLASK + POSTGRESQL ===

# Installation
# pip install flask psycopg2-binary flask-sqlalchemy

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/dbname'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# Modèle
class User(db.Model):
    __tablename__ = 'users'
    
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    age = db.Column(db.Integer)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'age': self.age,
            'created_at': self.created_at.isoformat()
        }

# Créer tables
with app.app_context():
    db.create_all()

# Routes
@app.route('/users', methods=['GET'])
def get_users():
    users = User.query.all()
    return jsonify([user.to_dict() for user in users])

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify(user.to_dict())

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    user = User(
        username=data['username'],
        email=data['email'],
        age=data.get('age')
    )
    db.session.add(user)
    db.session.commit()
    return jsonify(user.to_dict()), 201

@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    user = User.query.get_or_404(user_id)
    data = request.get_json()
    
    if 'username' in data:
        user.username = data['username']
    if 'email' in data:
        user.email = data['email']
    if 'age' in data:
        user.age = data['age']
    
    db.session.commit()
    return jsonify(user.to_dict())

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    return '', 204

if __name__ == '__main__':
    app.run(debug=True)


# === DJANGO ORM (Settings & Models) ===

# settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'mydb',
        'USER': 'myuser',
        'PASSWORD': 'mypassword',
        'HOST': 'localhost',
        'PORT': '5432',
    }
}

# models.py
from django.db import models

class User(models.Model):
    username = models.CharField(max_length=50, unique=True)
    email = models.EmailField(unique=True)
    age = models.IntegerField(null=True, blank=True)
    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    
    class Meta:
        db_table = 'users'
        ordering = ['-created_at']
        indexes = [
            models.Index(fields=['username']),
            models.Index(fields=['email']),
        ]
    
    def __str__(self):
        return self.username

class Order(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='orders')
    product = models.CharField(max_length=100)
    quantity = models.IntegerField()
    total = models.DecimalField(max_digits=10, decimal_places=2)
    created_at = models.DateTimeField(auto_now_add=True)
    
    class Meta:
        db_table = 'orders'

# Utilisation
from myapp.models import User, Order

# Créer
user = User.objects.create(username='john_doe', email='john@example.com', age=30)

# Récupérer
users = User.objects.all()
user = User.objects.get(id=1)
user = User.objects.filter(username='john_doe').first()

# Filter
users = User.objects.filter(age__gt=25)
users = User.objects.filter(username__icontains='john')
users = User.objects.filter(created_at__year=2024)

# Exclude
users = User.objects.exclude(age__lt=18)

# Q objects (complex queries)
from django.db.models import Q
users = User.objects.filter(Q(age__lt=20) | Q(age__gt=60))

# Mettre à jour
user.age = 31
user.save()

# Update sans charger
User.objects.filter(id=1).update(age=31)

# Supprimer
user.delete()
User.objects.filter(age__lt=18).delete()

# Relations
user = User.objects.get(id=1)
orders = user.orders.all()

# Prefetch (éviter N+1)
users = User.objects.prefetch_related('orders').all()

# Agrégation
from django.db.models import Count, Avg, Sum
User.objects.aggregate(avg_age=Avg('age'))
User.objects.annotate(order_count=Count('orders'))


# === CACHE AVEC REDIS ===

import redis
import json
from functools import wraps
from typing import Any, Callable

class RedisCache:
    def __init__(self, host='localhost', port=6379, db=0, ttl=3600):
        self.redis = redis.Redis(host=host, port=port, db=db, decode_responses=True)
        self.ttl = ttl
    
    def cache_result(self, key_prefix: str, ttl: int = None):
        """Décorateur pour cacher résultats de fonction"""
        def decorator(func: Callable) -> Callable:
            @wraps(func)
            def wrapper(*args, **kwargs) -> Any:
                # Créer clé unique
                key_parts = [key_prefix] + [str(arg) for arg in args]
                if kwargs:
                    key_parts.append(json.dumps(kwargs, sort_keys=True))
                cache_key = ':'.join(key_parts)
                
                # Vérifier cache
                cached = self.redis.get(cache_key)
                if cached:
                    return json.loads(cached)
                
                # Exécuter fonction
                result = func(*args, **kwargs)
                
                # Cacher résultat
                self.redis.setex(
                    cache_key,
                    ttl or self.ttl,
                    json.dumps(result)
                )
                
                return result
            return wrapper
        return decorator
    
    def invalidate(self, pattern: str):
        """Invalider cache par pattern"""
        for key in self.redis.scan_iter(match=pattern):
            self.redis.delete(key)

# Utilisation
cache = RedisCache()

@cache.cache_result('user', ttl=300)
def get_user(user_id: int):
    # Requête base de données coûteuse
    cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
    return dict(cursor.fetchone())

# Invalider cache
cache.invalidate('user:*')


# === SESSION STORE AVEC REDIS ===

import uuid
from datetime import timedelta

class SessionStore:
    def __init__(self, redis_client, prefix='session:', ttl=3600):
        self.redis = redis_client
        self.prefix = prefix
        self.ttl = ttl
    
    def create_session(self, user_id: int, data: dict = None) -> str:
        session_id = str(uuid.uuid4())
        session_key = f"{self.prefix}{session_id}"
        
        session_data = {
            'user_id': user_id,
            'created_at': str(datetime.utcnow()),
            **(data or {})
        }
        
        self.redis.setex(
            session_key,
            self.ttl,
            json.dumps(session_data)
        )
        
        return session_id
    
    def get_session(self, session_id: str) -> dict:
        session_key = f"{self.prefix}{session_id}"
        data = self.redis.get(session_key)
        return json.loads(data) if data else None
    
    def update_session(self, session_id: str, data: dict):
        session_key = f"{self.prefix}{session_id}"
        current = self.get_session(session_id)
        if current:
            current.update(data)
            self.redis.setex(session_key, self.ttl, json.dumps(current))
    
    def delete_session(self, session_id: str):
        session_key = f"{self.prefix}{session_id}"
        self.redis.delete(session_key)
    
    def refresh_session(self, session_id: str):
        session_key = f"{self.prefix}{session_id}"
        self.redis.expire(session_key, self.ttl)


# === DATA ACCESS LAYER (DAL) ===

from abc import ABC, abstractmethod
from typing import List, Optional, Generic, TypeVar

T = TypeVar('T')

class BaseRepository(ABC, Generic[T]):
    """Repository pattern abstrait"""
    
    @abstractmethod
    def create(self, entity: T) -> T:
        pass
    
    @abstractmethod
    def get(self, id: int) -> Optional[T]:
        pass
    
    @abstractmethod
    def get_all(self) -> List[T]:
        pass
    
    @abstractmethod
    def update(self, id: int, entity: T) -> Optional[T]:
        pass
    
    @abstractmethod
    def delete(self, id: int) -> bool:
        pass

class SQLiteUserRepository(BaseRepository):
    def __init__(self, db_path: str):
        self.db_path = db_path
    
    def create(self, user_data: dict) -> dict:
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute(
                'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
                (user_data['username'], user_data['email'], user_data.get('age'))
            )
            conn.commit()
            return self.get(cursor.lastrowid)
    
    def get(self, user_id: int) -> Optional[dict]:
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
            row = cursor.fetchone()
            return dict(row) if row else None
    
    def get_all(self) -> List[dict]:
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM users')
            return [dict(row) for row in cursor.fetchall()]
    
    def update(self, user_id: int, user_data: dict) -> Optional[dict]:
        with sqlite3.connect(self.db_path) as conn:
            fields = ', '.join(f'{k} = ?' for k in user_data.keys())
            values = list(user_data.values()) + [user_id]
            cursor = conn.cursor()
            cursor.execute(f'UPDATE users SET {fields} WHERE id = ?', values)
            conn.commit()
            return self.get(user_id) if cursor.rowcount > 0 else None
    
    def delete(self, user_id: int) -> bool:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
            conn.commit()
            return cursor.rowcount > 0


# === UNIT OF WORK PATTERN ===

class UnitOfWork:
    """Gestion transactionnelle multi-repository"""
    
    def __init__(self, db_session):
        self.session = db_session
        self._repositories = {}
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            self.rollback()
        else:
            self.commit()
        self.session.close()
    
    def commit(self):
        self.session.commit()
    
    def rollback(self):
        self.session.rollback()
    
    @property
    def users(self):
        if 'users' not in self._repositories:
            self._repositories['users'] = UserRepository(self.session)
        return self._repositories['users']
    
    @property
    def orders(self):
        if 'orders' not in self._repositories:
            self._repositories['orders'] = OrderRepository(self.session)
        return self._repositories['orders']

# Utilisation
with UnitOfWork(session) as uow:
    user = uow.users.create({'username': 'john', 'email': 'john@example.com'})
    order = uow.orders.create({'user_id': user['id'], 'product': 'Item'})
    # Auto-commit si pas d'exception


# === PAGINATION ===

class Paginator:
    def __init__(self, query, page: int = 1, per_page: int = 20):
        self.query = query
        self.page = max(1, page)
        self.per_page = per_page
    
    @property
    def offset(self):
        return (self.page - 1) * self.per_page
    
    def get_items(self):
        return self.query[self.offset:self.offset + self.per_page]
    
    @property
    def total(self):
        return len(self.query)
    
    @property
    def pages(self):
        return (self.total + self.per_page - 1) // self.per_page
    
    @property
    def has_prev(self):
        return self.page > 1
    
    @property
    def has_next(self):
        return self.page < self.pages
    
    def to_dict(self):
        return {
            'items': self.get_items(),
            'page': self.page,
            'per_page': self.per_page,
            'total': self.total,
            'pages': self.pages,
            'has_prev': self.has_prev,
            'has_next': self.has_next
        }

# SQLite pagination
def paginate_sqlite(cursor, query: str, page: int = 1, per_page: int = 20):
    offset = (page - 1) * per_page
    
    # Total count
    count_query = f"SELECT COUNT(*) FROM ({query})"
    cursor.execute(count_query)
    total = cursor.fetchone()[0]
    
    # Paginated results
    paginated_query = f"{query} LIMIT ? OFFSET ?"
    cursor.execute(paginated_query, (per_page, offset))
    items = cursor.fetchall()
    
    return {
        'items': items,
        'page': page,
        'per_page': per_page,
        'total': total,
        'pages': (total + per_page - 1) // per_page
    }

# SQLAlchemy pagination
from sqlalchemy import func

def paginate_sqlalchemy(query, page: int = 1, per_page: int = 20):
    total = query.count()
    items = query.limit(per_page).offset((page - 1) * per_page).all()
    
    return {
        'items': items,
        'page': page,
        'per_page': per_page,
        'total': total,
        'pages': (total + per_page - 1) // per_page
    }


# === FULL TEXT SEARCH ===

# SQLite FTS5
def setup_fts_sqlite(conn):
    conn.execute('''
        CREATE VIRTUAL TABLE IF NOT EXISTS users_fts 
        USING fts5(username, email, content)
    ''')
    
    # Trigger pour sync
    conn.execute('''
        CREATE TRIGGER IF NOT EXISTS users_fts_insert 
        AFTER INSERT ON users
        BEGIN
            INSERT INTO users_fts(rowid, username, email) 
            VALUES (new.id, new.username, new.email);
        END
    ''')

def search_fts(conn, query: str):
    cursor = conn.cursor()
    cursor.execute('''
        SELECT users.* 
        FROM users 
        JOIN users_fts ON users.id = users_fts.rowid 
        WHERE users_fts MATCH ?
    ''', (query,))
    return cursor.fetchall()

# PostgreSQL Full Text Search
def search_postgres(conn, query: str):
    cursor = conn.cursor()
    cursor.execute('''
        SELECT * FROM users 
        WHERE to_tsvector('english', username || ' ' || email) 
        @@ plainto_tsquery('english', %s)
    ''', (query,))
    return cursor.fetchall()

# Créer index GIN
# CREATE INDEX users_search_idx ON users 
# USING GIN (to_tsvector('english', username || ' ' || email));


# === SOFT DELETE ===

class SoftDeleteMixin:
    """Mixin pour soft delete (SQLAlchemy)"""
    deleted_at = Column(DateTime, nullable=True)
    
    def soft_delete(self):
        self.deleted_at = datetime.utcnow()
    
    def restore(self):
        self.deleted_at = None
    
    @property
    def is_deleted(self):
        return self.deleted_at is not None

# Query helper
def active_only(query):
    return query.filter_by(deleted_at=None)

# Utilisation
users = active_only(session.query(User)).all()


# === AUDIT LOG ===

class AuditLog(Base):
    __tablename__ = 'audit_logs'
    
    id = Column(Integer, primary_key=True)
    table_name = Column(String(100))
    record_id = Column(Integer)
    action = Column(String(20))  # INSERT, UPDATE, DELETE
    old_values = Column(Text)
    new_values = Column(Text)
    user_id = Column(Integer)
    timestamp = Column(DateTime, default=datetime.utcnow)

def log_change(table_name, record_id, action, old_values=None, new_values=None, user_id=None):
    log = AuditLog(
        table_name=table_name,
        record_id=record_id,
        action=action,
        old_values=json.dumps(old_values) if old_values else None,
        new_values=json.dumps(new_values) if new_values else None,
        user_id=user_id
    )
    session.add(log)
    session.commit()

# SQLAlchemy event listeners
from sqlalchemy import event

@event.listens_for(User, 'after_insert')
def log_user_insert(mapper, connection, target):
    log_change('users', target.id, 'INSERT', new_values={'username': target.username})

@event.listens_for(User, 'after_update')
def log_user_update(mapper, connection, target):
    log_change('users', target.id, 'UPDATE', new_values={'username': target.username})


[OK] OUTILS & BIBLIOTHÈQUES UTILES

# === ALEMBIC (Migrations) ===
# pip install alembic
# alembic init alembic
# alembic revision --autogenerate -m "message"
# alembic upgrade head

# === SQLALCHEMY-UTILS (Extensions) ===
# pip install sqlalchemy-utils
from sqlalchemy_utils import EmailType, URLType, UUIDType, ChoiceType

# === DATABASES (Async query builder) ===
# pip install databases
import databases
database = databases.Database('postgresql://user:pass@localhost/db')

# === DATASET (Simplification) ===
# pip install dataset
import dataset
db = dataset.connect('sqlite:///database.db')
table = db['users']
table.insert(dict(name='John', age=30))

# === RECORDS (SQL pour humains) ===
# pip install records
import records
db = records.Database('postgresql://...')
rows = db.query('SELECT * FROM users')

# === SQLPARSE (Parser SQL) ===
# pip install sqlparse
import sqlparse
formatted = sqlparse.format('select * from users', reindent=True)

# === PANDAS (Data analysis) ===
# pip install pandas
import pandas as pd
df = pd.read_sql_query('SELECT * FROM users', conn)
df.to_sql('users_backup', conn, if_exists='replace')

# === DBUTILS (Connection pooling) ===
# pip install dbutils
from dbutils.pooled_db import PooledDB
pool = PooledDB(sqlite3, maxconnections=10, database='database.db')

# === PYPIKA (Query builder) ===
# pip install pypika
from pypika import Query, Table
users = Table('users')
q = Query.from_(users).select('*').where(users.age > 25)


[OK] RESSOURCES & DOCUMENTATION

# Documentation officielle:
# - SQLite: https://docs.python.org/3/library/sqlite3.html
# - MySQL Connector: https://dev.mysql.com/doc/connector-python/
# - Psycopg2: https://www.psycopg.org/docs/
# - SQLAlchemy: https://docs.sqlalchemy.org/
# - PyMongo: https://pymongo.readthedocs.io/
# - Redis-py: https://redis-py.readthedocs.io/

# Tutoriels:
# - Real Python Database Tutorial
# - SQLAlchemy Tutorial
# - FastAPI with Databases

# Bonnes pratiques:
# - Use connection pooling in production
# - Always use parameterized queries
# - Implement proper error handling
# - Use transactions for data consistency
# - Index frequently queried columns
# - Monitor slow queries
# - Regular backups
# - Use ORM for complex applications
# - Cache frequently accessed data
# - Implement pagination for large datasets


[OK] COMPARAISON DES BASES DE DONNÉES

# SQLITE
# [OK] Zéro configuration
# [OK] Fichier unique portable
# [OK] Parfait pour développement/tests
# [OK] Excellent pour applications embarquées
# [X] Pas de concurrent writes
# [X] Limité pour applications web à fort trafic

# MYSQL
# [OK] Très populaire et mature
# [OK] Excellent pour web applications
# [OK] Bonnes performances lecture
# [OK] Réplication facile
# [X] Moins de fonctionnalités que PostgreSQL
# [X] Configuration plus complexe

# POSTGRESQL
# [OK] Le plus riche en fonctionnalités
# [OK] Excellent pour applications complexes
# [OK] JSONB natif
# [OK] Extensions puissantes
# [OK] Meilleure conformité SQL
# [X] Peut être plus lent que MySQL pour certains cas

# MONGODB
# [OK] Schéma flexible
# [OK] Excellent pour données non structurées
# [OK] Horizontal scaling facile
# [OK] Agrégation puissante
# [X] Pas de transactions multi-documents (avant 4.0)
# [X] Utilise plus de mémoire

# REDIS
# [OK] Extrêmement rapide (en mémoire)
# [OK] Excellent pour cache
# [OK] Pub/Sub natif
# [OK] Structures de données riches
# [X] Limité par RAM
# [X] Pas de requêtes complexes


# FIN DU CHEATSHEET