python-database-patterns
npx machina-cli add skill aiskillstore/marketplace/python-database-patterns --openclawFiles (1)
SKILL.md
4.7 KB
Python Database Patterns
SQLAlchemy 2.0 and database best practices.
SQLAlchemy 2.0 Basics
from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(String(255), unique=True)
is_active: Mapped[bool] = mapped_column(default=True)
# Create engine and tables
engine = create_engine("postgresql://user:pass@localhost/db")
Base.metadata.create_all(engine)
# Query with 2.0 style
with Session(engine) as session:
stmt = select(User).where(User.is_active == True)
users = session.execute(stmt).scalars().all()
Async SQLAlchemy
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy import select
# Async engine
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/db",
echo=False,
pool_size=5,
max_overflow=10,
)
# Session factory
async_session = async_sessionmaker(engine, expire_on_commit=False)
# Usage
async with async_session() as session:
result = await session.execute(select(User).where(User.id == 1))
user = result.scalar_one_or_none()
Model Relationships
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, Mapped, mapped_column
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
# One-to-many
posts: Mapped[list["Post"]] = relationship(back_populates="author")
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
# Many-to-one
author: Mapped["User"] = relationship(back_populates="posts")
Common Query Patterns
from sqlalchemy import select, and_, or_, func
# Basic select
stmt = select(User).where(User.is_active == True)
# Multiple conditions
stmt = select(User).where(
and_(
User.is_active == True,
User.age >= 18
)
)
# OR conditions
stmt = select(User).where(
or_(User.role == "admin", User.role == "moderator")
)
# Ordering and limiting
stmt = select(User).order_by(User.created_at.desc()).limit(10)
# Aggregates
stmt = select(func.count(User.id)).where(User.is_active == True)
# Joins
stmt = select(User, Post).join(Post, User.id == Post.author_id)
# Eager loading
from sqlalchemy.orm import selectinload
stmt = select(User).options(selectinload(User.posts))
FastAPI Integration
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session() as session:
yield session
DB = Annotated[AsyncSession, Depends(get_db)]
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: DB):
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(status_code=404)
return user
Quick Reference
| Operation | SQLAlchemy 2.0 Style |
|---|---|
| Select all | select(User) |
| Filter | .where(User.id == 1) |
| First | .scalar_one_or_none() |
| All | .scalars().all() |
| Count | select(func.count(User.id)) |
| Join | .join(Post) |
| Eager load | .options(selectinload(User.posts)) |
Additional Resources
./references/sqlalchemy-async.md- Async patterns, session management./references/connection-pooling.md- Pool configuration, health checks./references/transactions.md- Transaction patterns, isolation levels./references/migrations.md- Alembic setup, migration strategies
Assets
./assets/alembic.ini.template- Alembic configuration template
See Also
Prerequisites:
python-typing-patterns- Mapped types and annotationspython-async-patterns- Async database sessions
Related Skills:
python-fastapi-patterns- Dependency injection for DB sessionspython-pytest-patterns- Database fixtures and testing
Source
git clone https://github.com/aiskillstore/marketplace/blob/main/skills/0xdarkmatter/python-database-patterns/SKILL.mdView on GitHub Overview
This guide covers SQLAlchemy 2.0 patterns for Python apps, from ORM basics to async usage and migrations. It highlights practical patterns like repository and unit of work to keep data access clean, scalable, and easy to test, with FastAPI integration examples.
How This Skill Works
Using SQLAlchemy 2.0, the content demonstrates DeclarativeBase and Mapped columns, both in sync and async modes, with engine and session management. It shows common query patterns, relationships, and eager loading, plus how to wire in Alembic migrations for schema evolution and FastAPI endpoints for DB access.
When to Use It
- You are building a Python app that uses PostgreSQL and want to adopt SQLAlchemy 2.0 ORM patterns
- Your project requires asynchronous database access (asyncio) with FastAPI or similar frameworks
- You need to model clear relationships (one-to-many, many-to-one) between entities
- You plan to manage schema changes via migrations (Alembic) and track database evolution
- You want reusable data access patterns like repository and unit-of-work to decouple business logic from persistence
Quick Start
- Step 1: Install SQLAlchemy 2.0+ and an async driver (e.g., asyncpg or aiosqlite)
- Step 2: Define a DeclarativeBase, models with mapped_column, and relationships similar to the examples
- Step 3: Create an engine, session (or async_session) and run a simple select to query data
Best Practices
- Use DeclarativeBase and mapped_column for clear, type-safe models
- Prefer AsyncSession for async code; tune pool_size and max_overflow for performance
- Apply common query patterns with select, where, and eager loading (selectinload) to minimize queries
- Implement repository and unit of work patterns to encapsulate data access and transactions
- Integrate DB session lifecycle with FastAPI dependencies and handle commit/rollback correctly
Example Use Cases
- Define User and Post models with a one-to-many relationship and proper foreign keys
- Query asynchronously using AsyncSession and the async engine example
- Create a FastAPI endpoint that loads a user by ID with proper error handling
- Set up Alembic migrations to evolve the schema and generate migration scripts
- Demonstrate common patterns: filtering, ordering, joins, and aggregation across models
Frequently Asked Questions
Add this skill to your agents