Get the FREE Ultimate OpenClaw Setup Guide →

python-database-patterns

npx machina-cli add skill aiskillstore/marketplace/python-database-patterns --openclaw
Files (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

OperationSQLAlchemy 2.0 Style
Select allselect(User)
Filter.where(User.id == 1)
First.scalar_one_or_none()
All.scalars().all()
Countselect(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 annotations
  • python-async-patterns - Async database sessions

Related Skills:

  • python-fastapi-patterns - Dependency injection for DB sessions
  • python-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

  1. Step 1: Install SQLAlchemy 2.0+ and an async driver (e.g., asyncpg or aiosqlite)
  2. Step 2: Define a DeclarativeBase, models with mapped_column, and relationships similar to the examples
  3. 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
Sponsor this space

Reach thousands of developers