Python Academy · 课时

异步端点与数据库集成

Q: 「异步端点与数据库集成」这节课中我会学到什么？

编写异步路由处理器，并使用 SQLAlchemy 连接数据库。 你通过在浏览器中直接运行的动手代码来练习 Python Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

编写异步路由处理器，并使用 SQLAlchemy 连接数据库。

异步端点与数据库集成 是 CoddyKit 上的免费 Python Academy 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Python Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Python Academy 课程共包含 4 节课。

async def 端点

对于非阻塞 I/O，使用 async def 声明路由处理程序。对于 CPU 密集型或阻塞操作，使用 def（FastAPI 会在一个线程池中运行它们）。

from fastapi import FastAPI
import asyncio
app = FastAPI()

@app.get("/slow")
async def slow():
    await asyncio.sleep(1)   # non-blocking
    return {"done": True}

用于 PostgreSQL 的 asyncpg

asyncpg 是一个快速的异步 PostgreSQL 驱动程序。使用一个在多个请求之间共享的连接池。

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool("postgresql://user:pass@host/db")
    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, name FROM users")
    await pool.close()
    return rows

SQLAlchemy 异步功能

SQLAlchemy 1.4+ 配合 async_sessionmaker 可提供异步 ORM。将 AsyncSession 用作 FastAPI 依赖项。

# pip install sqlalchemy aiosqlite
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///db.sqlite")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

数据库依赖项

从 FastAPI 依赖项中生成会话，以便在请求完成后提交并关闭会话。

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

SQLAlchemy ORM 模型

使用 DeclarativeBase 定义 ORM 模型，并将其映射到数据库表。

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"

    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str]
    email: Mapped[str]

CRUD 操作

使用 SQLAlchemy 异步会话的方法实现基本的创建、读取、更新和删除操作。

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def create_user(db: AsyncSession, name: str, email: str):
    user = User(name=name, email=email)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def get_user(db: AsyncSession, user_id: int):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

使用 Alembic 进行迁移

使用 Alembic 以版本控制的方式管理数据库架构迁移。

# pip install alembic
# alembic init alembic
# Edit alembic/env.py to point to your models

# Generate a migration:
# alembic revision --autogenerate -m "add users table"

# Apply:
# alembic upgrade head

使用 httpx 进行异步 HTTP 调用

使用 httpx.AsyncClient 从异步 FastAPI 端点调用外部 API。

# pip install httpx
import httpx
from fastapi import FastAPI
app = FastAPI()

@app.get("/weather")
async def weather(city: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://wttr.in/{city}?format=j1")
        return r.json()

后台任务

使用 BackgroundTasks 在响应发送后运行任务，例如发送电子邮件或更新缓存。

from fastapi import FastAPI, BackgroundTasks
app = FastAPI()

def send_welcome_email(email: str):
    # blocking I/O — runs in background thread
    smtp_send(email, "Welcome!")

@app.post("/register")
def register(email: str, bg: BackgroundTasks):
    create_user(email)
    bg.add_task(send_welcome_email, email)
    return {"status": "registered"}

生命周期事件

使用 lifespan 上下文管理器（FastAPI 0.93+）运行启动和关闭代码，例如创建数据库连接池。

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await asyncpg.create_pool(DSN)
    yield
    await app.state.pool.close()

app = FastAPI(lifespan=lifespan)

使用 httpx 测试 FastAPI

使用带有 ASGITransport 的 httpx.AsyncClient 测试 FastAPI 端点，无需运行服务器。

import pytest, httpx
from fastapi import FastAPI
app = FastAPI()

@app.get("/ping")
async def ping(): return {"pong": True}

@pytest.mark.asyncio
async def test_ping():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        r = await client.get("/ping")
    assert r.json() == {"pong": True}

快速检查

对于 FastAPI 路由处理程序，什么时候应该使用 async def，什么时候应该使用 def？

回顾

将 async def 端点与异步数据库驱动程序（asyncpg、SQLAlchemy async）结合使用。将数据库会话作为 FastAPI 依赖项进行管理。使用 BackgroundTasks 处理响应发送后的任务，并使用 lifespan 上下文管理器处理启动和关闭。

常见问题解答

「异步端点与数据库集成」课时是免费的吗？

是的 — 「异步端点与数据库集成」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Python Academy 课程的其余内容，请升级到 CoddyKit PRO。 Python Academy 课程共包含 4 节课。

「异步端点与数据库集成」这节课中我会学到什么？

编写异步路由处理器，并使用 SQLAlchemy 连接数据库。你通过在浏览器中直接运行的动手代码来练习 Python Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Python Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Python Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「异步端点与数据库集成」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Python Academy 课中编写并运行代码吗？

能。每节 Python Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 Python Academy