tech
Defaultdeesfrzh

SQLAlchemy

Toolkit SQL Python et ORM - requetes flexibles, relations, support async, multi-BDD

TL;DR

Quoi : Boîte à outils SQL Python et Object-Relational Mapper.

Pourquoi : Flexible, puissant, supporte SQL brut et ORM, agnostique de base de données.

Quick Start

Installer :

pip install sqlalchemy

Connecter et interroger :

from sqlalchemy import create_engine, text

engine = create_engine('sqlite:///example.db')

with engine.connect() as conn:
    result = conn.execute(text("SELECT 'Hello, SQLAlchemy!'"))
    print(result.all())

Cheatsheet

ConceptDescription
EngineConnexion base de données
SessionUnité de travail
ModelMapping de table
ColumnDéfinition de champ
relationshipRelations entre modèles
QueryRécupération de données

Gotchas

Model definition (ORM)

from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True)

    posts = relationship('Post', back_populates='author')

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    author_id = Column(Integer, ForeignKey('users.id'))

    author = relationship('User', back_populates='posts')

# Create tables
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)

CRUD operations

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

# Create
user = User(name='John', email='[email protected]')
session.add(user)
session.commit()

# Read
users = session.query(User).all()
user = session.query(User).filter_by(id=1).first()
user = session.query(User).filter(User.name == 'John').first()

# Update
user.name = 'Johnny'
session.commit()

# Delete
session.delete(user)
session.commit()

Query filtering

from sqlalchemy import and_, or_

# Basic filters
users = session.query(User).filter(User.name == 'John').all()
users = session.query(User).filter(User.name.like('%John%')).all()
users = session.query(User).filter(User.id.in_([1, 2, 3])).all()

# Multiple conditions
users = session.query(User).filter(
    and_(
        User.name == 'John',
        User.email.like('%@example.com')
    )
).all()

# Ordering and limiting
users = session.query(User).order_by(User.name).limit(10).all()

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

Relationships

# Query with relationship
user = session.query(User).filter_by(id=1).first()
print(user.posts)  # Lazy load posts

# Eager loading
from sqlalchemy.orm import joinedload

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

# Create with relationship
post = Post(title='Hello', author=user)
session.add(post)
session.commit()

SQLAlchemy 2.0 style

from sqlalchemy import select
from sqlalchemy.orm import Session

with Session(engine) as session:
    # Select
    stmt = select(User).where(User.name == 'John')
    users = session.scalars(stmt).all()

    # Insert
    user = User(name='Jane', email='[email protected]')
    session.add(user)
    session.commit()

Async support

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine('postgresql+asyncpg://user:pass@localhost/db')
async_session = sessionmaker(engine, class_=AsyncSession)

async with async_session() as session:
    result = await session.execute(select(User))
    users = result.scalars().all()

Next Steps