Litestar 构建 Web API 入门指南

Litestar 是 Python 新一代高性能异步 API 框架:类型安全、依赖注入、自动校验。本文以博客 API 为例讲解 SQLAlchemy 集成与完整 CRUD 实现。

最佳实践
并发任务的多路径协同插画

直接回答:Litestar 是 Python 异步 API 框架的后起之秀:类型注解驱动、依赖注入强大、性能优化内置,开发体验对标 FastAPI 且在架构分层上更有主见。 想尝试 FastAPI 之外的现代选择,它值得认真评估。

Litestar 的定位

Litestar(前身 Starlite)在 FastAPI 验证过的方向上更进一步:DTO(数据传输对象)一等公民、插件化 ORM 集成、精细化缓存与 OpenAPI 生成。社区虽年轻,但设计与代码质量口碑俱佳。

搭建项目

pip install 'litestar[standard,sqlalchemy]>=2.24,<3' sqlalchemy aiosqlite

配置数据库

Litestar 官方插件接管 SQLAlchemy 会话生命周期:

from litestar import Litestar
from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin

from litestar.di import NamedDependency
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

模型与 Schema

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): ...

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    body: Mapped[str]

config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///blog.db",
    metadata=Base.metadata, create_all=True,
    session_dependency_key="session",
)
plugin = SQLAlchemyPlugin(config=config)

Litestar 的分层哲学:数据库模型是数据库模型,API 契约是 API 契约,DTO 在两者之间做显式翻译——比"一个模型打天下"更能扛住需求演进。

创建与查询

from litestar import get, post, put, delete
from litestar.exceptions import NotFoundException
from dataclasses import dataclass

@dataclass
class PostCreate:
    title: str
    body: str

PostUpdate = PostCreate  # PUT 要求完整输入

@dataclass
class PostView:
    id: int
    title: str
    body: str

def as_view(post: Post) -> PostView:
    return PostView(id=post.id, title=post.title, body=post.body)

@post("/posts")
async def create_post(data: PostCreate, session: NamedDependency[AsyncSession]) -> PostView:
    post = Post(title=data.title, body=data.body)
    session.add(post)
    await session.commit()
    await session.refresh(post)
    return as_view(post)

@get("/posts")
async def list_posts(session: NamedDependency[AsyncSession]) -> list[PostView]:
    result = await session.execute(select(Post).order_by(Post.id.desc()).limit(100))
    return [as_view(post) for post in result.scalars()]

@get("/posts/{pid:int}")
async def get_post(pid: int, session: NamedDependency[AsyncSession]) -> PostView:
    post = await session.get(Post, pid)
    if not post:
        raise NotFoundException("文章不存在")
    return as_view(post)

异步端到端:路由是 async,SQLAlchemy 会话是 async——IO 等待期间事件循环服务其他请求。

更新与删除

@put("/posts/{pid:int}")
async def update_post(pid: int, data: PostUpdate, session: NamedDependency[AsyncSession]) -> PostView:
    post = await session.get(Post, pid)
    if not post:
        raise NotFoundException()
    post.title = data.title
    post.body = data.body
    await session.commit()
    await session.refresh(post)
    return as_view(post)

@delete("/posts/{pid:int}")
async def delete_post(pid: int, session: NamedDependency[AsyncSession]) -> None:
    post = await session.get(Post, pid)
    if not post:
        raise NotFoundException()
    await session.delete(post)
    await session.commit()

最后组装 app = Litestar(plugins=[plugin], route_handlers=[create_post, list_posts, get_post, update_post, delete_post]),保存为 app.py 后运行 litestar --app app:app run。以上用显式 dataclass 输入和响应转换,不把 DTO 类当返回数据类型;生产需增加字段约束、认证、对象权限和分页,create_all 只用于本地建表,生产使用迁移。

常见问题(FAQ)

Q:Litestar 和 FastAPI 怎么选?
A:FastAPI 生态与资料更厚;Litestar 的 DTO 分层、DI 体系更"工程化",适合中大型 API 项目。小项目两边都行,大项目值得给 Litestar 一次评估。

Q:Litestar 支持模板与静态文件吗?
A:支持,但它的重心在 API。全栈服务端渲染场景 Django/Flask 更顺手。

Q:异步 SQLAlchemy 有什么坑?
A:懒加载在异步下要显式 selectinload;会话不能跨请求共享(Litestar 插件已管好生命周期)。记住"异步里所有 IO 都要 await"。

官方参考

本文基于官方文档整理,未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全;局部片段不等同于完整生产应用。

延伸阅读

获取专属方案

联系我们

加入社区

微信扫码
加入官方交流群

立即体验

在线开通,按量计费,真正的云服务!

立即开始

选择观测云版本

代码托管平台