python-database-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Python Database Patterns

Python 数据库模式

SQLAlchemy 2.0 and database best practices.

SQLAlchemy 2.0 与数据库最佳实践。

SQLAlchemy 2.0 Basics

SQLAlchemy 2.0 基础

python

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)

python

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()

undefined

with Session(engine) as session: stmt = select(User).where(User.is_active == True) users = session.execute(stmt).scalars().all()

undefined

Async SQLAlchemy

异步 SQLAlchemy

python

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy import select

python

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()

undefined

async with async_session() as session: result = await session.execute(select(User).where(User.id == 1)) user = result.scalar_one_or_none()

undefined

Model Relationships

模型关系

python

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")

python

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

常见查询模式

python

from sqlalchemy import select, and_, or_, func

python

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))

undefined

from sqlalchemy.orm import selectinload stmt = select(User).options(selectinload(User.posts))

undefined

FastAPI Integration

FastAPI 集成

python

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

python

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))`

操作	SQLAlchemy 2.0 风格
查询全部	`select(User)`
过滤	`.where(User.id == 1)`
获取单个或None	`.scalar_one_or_none()`
获取全部	`.scalars().all()`
计数	`select(func.count(User.id))`
连接	`.join(Post)`
预加载	`.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

```
./references/sqlalchemy-async.md
```
- 异步模式、会话管理
```
./references/connection-pooling.md
```
- 连接池配置、健康检查
```
./references/transactions.md
```
- 事务模式、隔离级别
```
./references/migrations.md
```
- Alembic 配置、迁移策略

Assets

资源文件

```
./assets/alembic.ini.template
```
- Alembic configuration template

```
./assets/alembic.ini.template
```
- Alembic 配置模板

另请参阅

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

前置要求:

```
python-typing-patterns
```
- 映射类型与注解
```
python-async-patterns
```
- 异步数据库会话

相关技能:

```
python-fastapi-patterns
```
- 数据库会话的依赖注入
```
python-pytest-patterns
```
- 数据库 fixtures 与测试

python-database-patterns

Original

Translation

Python Database Patterns

Python 数据库模式

SQLAlchemy 2.0 Basics

SQLAlchemy 2.0 基础

Create engine and tables

Create engine and tables

Query with 2.0 style

Query with 2.0 style

Async SQLAlchemy

异步 SQLAlchemy

Async engine

Async engine

Session factory

Session factory

Usage

Usage

Model Relationships

模型关系

Common Query Patterns

常见查询模式

Basic select

Basic select

Multiple conditions

Multiple conditions

OR conditions

OR conditions

Ordering and limiting

Ordering and limiting

Aggregates

Aggregates

Joins

Joins

Eager loading

Eager loading

FastAPI Integration

FastAPI 集成

Quick Reference

快速参考

Additional Resources

额外资源

Assets

资源文件

See Also

另请参阅