Sqlalchemy Postgres
Ship async FastAPI routes and repositories on PostgreSQL using SQLAlchemy 2.x session and dependency patterns without reinventing boilerplate each project.
Overview
SQLAlchemy Postgres is an agent skill for the Build phase that teaches async SQLAlchemy session, repository, and FastAPI lifespan patterns for PostgreSQL-backed APIs.
Install
npx skills add https://github.com/cfircoo/claude-code-toolkit --skill sqlalchemy-postgresWhat is this skill?
- FastAPI `get_db` async generator with rollback on exception
- `DBSession` and repository `Depends` type aliases for clean route signatures
- Lifespan startup connection check and engine dispose on shutdown
- Async session factory patterns for Postgres-backed services
Adoption & trust: 1 installs on skills.sh; 17 GitHub stars; 3/3 security scanners passed (skills.sh audits); trending (+100% hot-view momentum).
What problem does it solve?
You are building a FastAPI service on Postgres but keep getting async session scope, rollback, and dependency injection wrong across routes.
Who is it for?
Solo builders shipping a Python API who want copy-ready FastAPI + async SQLAlchemy structure without reading the full ORM docs end to end.
Skip if: Frontend-only projects, sync Django stacks, or teams that already standardize on a different ORM layer with locked internal templates.
When should I use this skill?
When implementing or refactoring async database access in FastAPI applications backed by PostgreSQL.
What do I get? / Deliverables
Your agent applies consistent async session factories, DBSession aliases, repository Depends, and lifespan hooks so API code matches maintainable SQLAlchemy 2.x practice.
- Session dependency module
- Repository Depends wiring
- Lifespan database startup/shutdown hooks
Recommended Skills
Journey fit
How it compares
Opinionated FastAPI dependency templates—not a raw SQL or migration generator skill.
Common Questions / FAQ
Who is sqlalchemy-postgres for?
Indie developers and small teams building async Python APIs on PostgreSQL who want their coding agent to follow the same session and repository patterns every time.
When should I use sqlalchemy-postgres?
Use it in Build → backend when scaffolding routes, repositories, and app lifespan for a new FastAPI service or refactoring messy database access in an existing Postgres project.
Is sqlalchemy-postgres safe to install?
It is procedural code guidance; review the Security Audits panel on this page and avoid embedding production credentials in prompts—use env vars and least-privilege DB roles.
SKILL.md
READMESKILL.md - Sqlalchemy Postgres
# Async SQLAlchemy Patterns <fastapi_integration> ## FastAPI Integration ### Session Dependency ```python from typing import Annotated, AsyncGenerator from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession async def get_db() -> AsyncGenerator[AsyncSession, None]: async with async_session_factory() as session: try: yield session except Exception: await session.rollback() raise # Type alias for cleaner signatures DBSession = Annotated[AsyncSession, Depends(get_db)] # Usage in routes @router.get("/users/{user_id}") async def get_user(user_id: int, db: DBSession): user = await db.execute(select(User).where(User.id == user_id)) return user.scalar_one_or_none() ``` ### Repository Dependency ```python def get_user_repo(session: DBSession) -> UserRepository: return UserRepository(session) UserRepo = Annotated[UserRepository, Depends(get_user_repo)] @router.get("/users") async def list_users(repo: UserRepo): return await repo.get_multi() ``` ### Lifespan Event for Connection ```python from contextlib import asynccontextmanager from fastapi import FastAPI @asynccontextmanager async def lifespan(app: FastAPI): # Startup: verify connection async with engine.begin() as conn: await conn.execute(text("SELECT 1")) yield # Shutdown: dispose engine await engine.dispose() app = FastAPI(lifespan=lifespan) ``` </fastapi_integration> <async_session_patterns> ## Async Session Patterns ### Context Manager Pattern ```python async def create_user(email: str) -> User: async with async_session_factory() as session: user = User(email=email) session.add(user) await session.commit() await session.refresh(user) return user ``` ### Manual Transaction Control ```python async def transfer_funds(from_id: int, to_id: int, amount: float): async with async_session_factory() as session: async with session.begin(): # Both operations in same transaction from_account = await session.get(Account, from_id) to_account = await session.get(Account, to_id) if from_account.balance < amount: raise ValueError("Insufficient funds") from_account.balance -= amount to_account.balance += amount # Auto-commit on exit ``` ### Nested Transactions (Savepoints) ```python async def complex_operation(): async with async_session_factory() as session: async with session.begin(): session.add(user) try: async with session.begin_nested(): session.add(risky_operation) except Exception: # Only risky_operation rolled back pass # user still committed ``` </async_session_patterns> <eager_loading_async> ## Eager Loading in Async **Critical**: Lazy loading doesn't work with async. Always use eager loading. ### selectinload (Best for Collections) ```python # Loads related collection with SELECT ... IN (...) stmt = select(User).options(selectinload(User.posts)) result = await session.execute(stmt) users = result.scalars().all() for user in users: print(user.posts) # Already loaded, no additional query ``` ### joinedload (Best for Single Relations) ```python # Loads related object with JOIN stmt = select(User).options(joinedload(User.organization)) result = await session.execute(stmt) users = result.scalars().unique().all() # Note: unique() needed with joins ``` ### Nested Eager Loading ```python stmt = select(User).options( selectinload(User.posts).selectinload(Post.comments), joinedload(User.organization), ) ``` ### contains_eager (With Explicit Join) ```python from sqlalchemy.orm import contains_eager stmt = ( select(User) .join(User.posts) .where(Post.is_published == True) .options(contains_eager(User.posts)) ) ``` </eager_loading_async> <concu