SQLAlchemy 基础教程

下面给你整理一份真正适用于生产环境的、基于 SQLAlchemy 2.0 语法的完整教程（含 ORM + Core + Async + 会话管理 + 查询模型），覆盖你日常开发 FastAPI、PostgreSQL、TimescaleDB、RBAC 的全部常用操作。

全部代码均为 SQLAlchemy 2.0 官方推荐写法（如 Session(engine)、select()、async with AsyncSession()、mappings() 等）。

✅ SQLAlchemy 2.0 完整教程（生产级别）

🍃 安装
🚀 创建 Engine
🧱 定义 ORM 模型
📀 创建表
🧩 Session（同步 & 异步）
🔍 ORM 查询（select / where / order）
✏️ 插入（ORM / 原生 SQL）
🛠 更新、删除
📌 Result 解析（scalars/mappings/fetchall）
🔗 JOIN 查询
📂 Core 风格的执行
⚡ Async SQLAlchemy 用法
🧪 事务管理
🔒 最佳实践（你的生产项目应该这样写）

1. 安装

pip install sqlalchemy psycopg2-binary asyncpg

2. 创建 Engine（同步 & 异步）

同步 engine

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg2://user:pass@localhost/dbname",
    future=True,
)

异步 engine（asyncpg）

from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/dbname",
    future=True,
)

3. 定义 ORM 模型（2.0 写法）

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import Column, Integer, String

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True)
    role = Column(String(20))

4. 创建表

Base.metadata.create_all(engine)

5. Session（推荐 2.0 style）

同步

from sqlalchemy.orm import Session

with Session(engine) as session:
    ...

异步

from sqlalchemy.ext.asyncio import AsyncSession

async with AsyncSession(async_engine) as session:
    ...

6. ORM 查询（select + where）

from sqlalchemy import select

stmt = select(User).where(User.role == "admin")

with Session(engine) as session:
    result = session.execute(stmt)
    users = result.scalars().all()

7. 插入

ORM 插入（推荐）

user = User(username="martin", role="admin")

with Session(engine) as s:
    s.add(user)
    s.commit()

原生 SQL 插入

from sqlalchemy import text

with engine.connect() as conn:
    conn.execute(text("INSERT INTO users (username) VALUES (:u)"), {"u": "martin"})
    conn.commit()

8. 更新 & 删除

更新（ORM）

stmt = (
    update(User)
    .where(User.id == 1)
    .values(role="superadmin")
)

with Session(engine) as s:
    s.execute(stmt)
    s.commit()

删除（ORM）

stmt = delete(User).where(User.id == 1)

with Session(engine) as s:
    s.execute(stmt)
    s.commit()

9. Result 解析（SQLAlchemy 2.0 最重要部分）

返回 ORM 对象

result.scalars().all()

字典格式（最常用）

result.mappings().all()

元组格式

result.fetchall()

10. JOIN 查询

stmt = (
    select(User.username, Role.name)
    .select_from(User)
    .join(Role, User.role_id == Role.id)
)

11. 使用 Core 执行 SQL（连接驱动）

with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users WHERE id=:id"), {"id": 1})
    rows = result.mappings().all()

12. Async SQLAlchemy 快速示例

async def get_users():
    async with AsyncSession(async_engine) as session:
        result = await session.execute(select(User))
        return result.scalars().all()

13. 事务管理（自动提交/回滚）

with Session(engine) as session, session.begin():
    session.execute(...)

14. SQLAlchemy 生产最佳实践（⭐ 强烈建议你采用）

engine 是单例
封装 SessionLocal
封装 CRUD
统一结果格式
避免连接泄漏