Models, Sessions & Queries

Declarative Models

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

Base = declarative_base()

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True)  # DB-level constraint — rejects duplicates
    posts = relationship('Post', back_populates='author')

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    author_id = Column(Integer, ForeignKey('authors.id'))
    author = relationship('Author', back_populates='posts')

engine = create_engine('postgresql://user:pass@localhost/mydb')  # manages a
Base.metadata.create_all(engine)                                  # connection pool
Session = sessionmaker(bind=engine)

Session Lifecycle

session = Session()

try:
    author = Author(name='Alice', email='alice@example.com')
    session.add(author)
    session.flush()   # sends SQL now, so author.id is populated — but the
                      # transaction is still open, not yet permanent
    print(author.id)

    post = Post(title='Hello World', author_id=author.id)
    session.add(post)
    session.commit()  # flush + actually commit — now permanently visible
except Exception:
    session.rollback()  # discard everything since the last commit
    raise
finally:
    session.close()

Querying

from sqlalchemy import select

# Classic style
adults = session.query(User).filter(User.age > 18).order_by(User.name).all()

# 2.0-style, unified across ORM and Core
adults = session.execute(
    select(User).where(User.age > 18).order_by(User.name)
).scalars().all()

# Unique constraint violations surface as IntegrityError — don't rely on
# a check-then-insert alone, which races under concurrent requests
from sqlalchemy.exc import IntegrityError
try:
    session.add(User(email='taken@example.com'))
    session.commit()
except IntegrityError:
    session.rollback()
    raise ValueError('email already taken')