skb3002/agenthub/sqlalchemy-expert

Claude Code and agent skill by skb3002/agenthub

Trust tier 3
  • community
Source check pending

Checking that the SKILL.md file is still in the repository.

SQLAlchemy 2.0 async patterns, session management, query optimization, and Alembic migrations. Use when designing Python data access layers with SQLAlchemy.

Install

npx vskill@latest install skb3002/agenthub/sqlalchemy-expert

SKILL.md

SQLAlchemy 2.0 Expert

Modern async-first SQLAlchemy patterns for production Python services.
Learn to THINK about the session lifecycle, not copy query strings.

⚠️ How to Use This Skill

  • ASK whether the service is async or sync before choosing the API
  • Default to SQLAlchemy 2.0 style (select(), Mapped[], typed) — NOT the legacy 1.4 Query API
  • If raw SQL fits better, use it — an ORM isn't mandatory

1. When to Use SQLAlchemy vs Raw SQL

Decision Matrix
Use SQLAlchemy when…Use Raw SQL when…
Multi-table ORM relationshipsComplex analytics / window functions
Migrations via AlembicBulk ETL / data pipelines
Schema-driven appSingle known query, perf critical
Team wants type safetyDB-specific features (CTE, LATERAL)
Hybrid Pattern (Recommended)
ORM for CRUD + relationships.
Raw SQL (via session.execute(text(...))) for reports and hot paths.
Both share the same session + transaction.

2. Async Setup (SQLAlchemy 2.0)

Engine & Session
python
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
engine = create_async_engine(
settings.database_url, # postgresql+asyncpg://... or mysql+aiomysql://...
pool_size=10,
max_overflow=20,
pool_pre_ping=True, # detect stale connections
echo=settings.debug,
)
AsyncSessionLocal = async_sessionmaker(
engine,
expire_on_commit=False, # keep objects usable after commit
)
Driver Matrix
DBSync DriverAsync DriverDialect
PostgreSQLpsycopg2 / psycopg3asyncpgpostgresql+asyncpg
MySQLPyMySQLaiomysqlmysql+aiomysql
SQLitesqlite3aiosqlitesqlite+aiosqlite
SQL Serverpyodbcaioodbcmssql+aioodbc

3. Session Management

The Golden Rule

One session per request, one transaction per logical unit of work.

FastAPI Dependency
python
async def get_db() -> AsyncIterator[AsyncSession]:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
Transactions
PatternUse
async with session.begin():Explicit transaction block
async with session.begin_nested():Savepoint (nested transaction)
Auto-commit on dep yieldDefault request lifecycle

4. Modern Model Style (SQLAlchemy 2.0)

Typed Mapped[] Syntax
python
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, ForeignKey
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
posts: Mapped[list["Post"]] = relationship(back_populates="author")
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
author: Mapped[User] = relationship(back_populates="posts")
Benefits Over 1.4 Style
  • mypy catches type errors at declaration
  • No more Column(Integer, primary_key=True) — types come from annotations
  • Relationship types flow through queries

5. Querying (2.0 API)

SELECT — always use select()
python
from sqlalchemy import select
# One
stmt = select(User).where(User.email == email)
user = (await session.execute(stmt)).scalar_one_or_none()
# Many
stmt = select(User).where(User.active.is_(True)).order_by(User.created_at.desc())
users = (await session.execute(stmt)).scalars().all()
INSERT / UPDATE / DELETE
python
from sqlalchemy import insert, update, delete
# Insert (returning)
stmt = insert(User).values(email=email).returning(User.id)
new_id = (await session.execute(stmt)).scalar_one()
# Update
await session.execute(
update(User).where(User.id == uid).values(last_seen=now())
)
# Delete
await session.execute(delete(User).where(User.id == uid))
Eager Loading — the N+1 Fix
PatternWhen
selectinload(User.posts)Many-to-many, one-to-many (issues 2nd query)
joinedload(User.profile)One-to-one, one-to-few (single JOIN)
.options(contains_eager(...))Manual JOIN with explicit load
python
stmt = select(User).options(selectinload(User.posts)).where(User.active)

6. Relationships

Cardinality Map
RelationshipPattern
One-to-onerelationship(..., uselist=False) + unique FK
One-to-manyFK on child, relationship on parent
Many-to-manyAssociation table with two FKs
Self-referentialremote_side=[id] for tree structures
Cascades — Be Explicit
python
posts: Mapped[list[Post]] = relationship(
back_populates="author",
cascade="all, delete-orphan", # delete posts when user deleted
passive_deletes=True, # let DB ON DELETE CASCADE do the work
)

7. Alembic Migrations

Setup
bash
alembic init alembic
# Edit alembic.ini + env.py for async + target_metadata = Base.metadata
Common Flow
bash
alembic revision --autogenerate -m "add users table"
# REVIEW the generated file — autogenerate misses enum renames, index changes
alembic upgrade head
Migration Hygiene
RuleWhy
Number sequentially (001_, 002_, …)Predictable ordering in teams
Small migrationsEasier to review + roll back
Never edit a merged migrationWrite a new one
Review autogenerated outputAutogen is a draft, not truth
Test rollbackalembic downgrade -1 should work
Zero-Downtime Strategy
Adding column: nullable → deploy → backfill → NOT NULL
Removing column: stop using in app → deploy → drop column
Renaming column: add new → migrate data → deploy → drop old
Adding index: use CREATE INDEX CONCURRENTLY (Postgres)

8. Performance

Connection Pool Tuning
SettingDefaultTypical Prod
pool_size510–20
max_overflow1020
pool_timeout3030
pool_recycle-11800 (30 min)
pool_pre_pingFalseTrue
Query Optimization
Steps:
1. Identify slow query (APM, echo=True in dev)
2. Run EXPLAIN ANALYZE
3. Check index usage
4. Fix N+1 with selectinload / joinedload
5. Profile in prod with sample queries
Batching
python
# Bad: N round-trips
for email in emails:
await session.execute(insert(User).values(email=email))
# Good: one round-trip
await session.execute(insert(User), [{"email": e} for e in emails])

9. Testing

Transactional Fixtures
python
@pytest.fixture
async def db_session():
async with engine.connect() as conn:
trans = await conn.begin()
session = AsyncSession(bind=conn)
try:
yield session
finally:
await session.close()
await trans.rollback() # rollback → test isolation

Every test starts from a clean DB state without re-creating the schema.

Don't Mock the ORM

Use a real SQLite (sqlite+aiosqlite:///:memory:) or Postgres test container. Mocking the session is fragile and hides bugs.


10. Anti-Patterns

❌ DON'T
  • Mix sync and async drivers in the same app
  • Use session.query() (legacy 1.4 API)
  • Rely on autocommit=True (removed in 2.0)
  • Share a session across threads or async tasks
  • Use expire_on_commit=True (default) with long-lived objects
  • Skip pool_pre_ping in production
  • Forget to review Alembic autogenerated migrations
✅ DO
  • Use select(), insert(), update(), delete() functions
  • One session per request, commit once at boundary
  • Eager-load relationships with selectinload / joinedload
  • Keep migrations small and reversible
  • Test with real DB, not mocks
  • Type models with Mapped[]

11. SQLAlchemy vs Alternatives

LibraryBest ForWhy pick it
SQLAlchemy 2.0Production apps, complex schemasMature, typed, async, any DB
SQLModelFastAPI-heavy, Pydantic-firstWraps SQLAlchemy with Pydantic models
Tortoise ORMAsync-native, Django-like APISimpler API, smaller ecosystem
PiccoloAsync + ORM + admin UIBuilt-in admin, smaller scope
asyncpg directHot paths, known queriesNo ORM overhead, raw speed

Remember: SQLAlchemy is a tool for managing sessions and transactions, not just queries. Master the session lifecycle and the rest follows.

Security

  • Security scanPassed, 100 of 100Pattern checks on the skill's files.
  • Model reviewPassed

Quality evaluation

  • Task evaluationNot runNo evaluation yet. Run one in Skill Studio.

Source

Popularity

Repository stars
3
7-day trend score
7

Stars and forks are for the whole GitHub repository, not this skill alone. Not refreshed yet.

Works with

Any agent vskill installs to: 56 targets in all. These 39 have a page on this site.

  • Claude Code
  • Cursor
  • GitHub Copilot
  • Windsurf
  • Codex
  • Gemini CLI
  • Cline
  • Amp
  • Roo Code
  • Goose
  • Continue
  • Kilo Code
  • Junie
  • Trae
  • OpenHands
  • Qwen Code
  • AI Agent
  • Tabnine Chat
  • Sourcegraph Cody
  • Amazon Q
  • Firebase Genkit
  • Sweep
  • Devin
  • Phind
  • Mentat
  • Claude Desktop
  • Aider
  • GPT Pilot
  • Plandex
  • bolt.new
  • v0
  • Replit Agent
  • Anse
  • TypingMind
  • BetterChatGPT
  • LibreChat
  • Warp AI
  • Pieces
  • Fig AI
Embed badge
vskill VERIFIED badge
[![vskill VERIFIED](https://verified-skill.com/api/v1/skills/skb3002/agenthub/sqlalchemy-expert/badge)](https://verified-skill.com/skills/skb3002/agenthub/sqlalchemy-expert)

Something wrong with this skill or its page? Tell us.