跳转至

Litestar Python API 框架

为什么要学 Litestar

Litestar(原 Starlite)是一个高性能的 Python ASGI 框架,定位在 FastAPI 和 Django 之间。它提供了类型驱动的 API 设计、内置依赖注入、OpenAPI 文档生成、数据验证,同时还有 FastAPI 缺少的功能:内置 ORM 集成(SQLAlchemy/Tortoise)、DTO 系统、WebSocket 支持、后台任务队列等。对于需要更全面框架特性的 Python API 项目来说,Litestar 是值得考虑的选择。


核心概念

概念白话解释用途
Route Handler路由处理器处理 HTTP 请求的函数/类
Controller控制器组织路由的类
DTO数据传输对象控制输入输出的数据形状
Guard守卫权限检查
Middleware中间件请求/响应处理链
Dependency依赖依赖注入系统

安装配置

pip install litestar[standard]
# standard 包含:uvicorn, jinja2, pydantic 等

快速上手

最简 API

from litestar import Litestar, get

@get("/")
async def index() -> dict[str, str]:
    return {"message": "Hello, Litestar!"}

@get("/users/{user_id:int}")
async def get_user(user_id: int) -> dict:
    return {"id": user_id, "name": "Alice"}

app = Litestar([index, get_user])
litestar run  # 或 uvicorn main:app

Controller 组织

from litestar import Controller, get, post, put, delete
from pydantic import BaseModel

class UserModel(BaseModel):
    id: int
    name: str
    email: str

class UserCreate(BaseModel):
    name: str
    email: str

class UserController(Controller):
    path = "/users"

    @get()
    async def list_users(self) -> list[UserModel]:
        return [UserModel(id=1, name="Alice", email="alice@example.com")]

    @post()
    async def create_user(self, data: UserCreate) -> UserModel:
        return UserModel(id=1, **data.model_dump())

    @get("/{user_id:int}")
    async def get_user(self, user_id: int) -> UserModel:
        return UserModel(id=user_id, name="Alice", email="alice@example.com")

    @delete("/{user_id:int}")
    async def delete_user(self, user_id: int) -> None:
        return None

app = Litestar(route_handlers=[UserController])

依赖注入

from litestar import Litestar, get
from litestar.di import Provide

async def get_db_session():
    session = create_session()
    try:
        yield session
    finally:
        session.close()

@get("/users", dependencies={"session": Provide(get_db_session)})
async def get_users(session) -> list[dict]:
    return session.query(User).all()

进阶用法

Guard 权限控制

from litestar import Litestar, get
from litestar.connection import ASGIConnection
from litestar.handlers import BaseRouteHandler
from litestar.exceptions import NotAuthorizedException

def auth_guard(connection: ASGIConnection, _: BaseRouteHandler) -> None:
    if not connection.headers.get("Authorization"):
        raise NotAuthorizedException("Missing auth token")

@get("/protected", guards=[auth_guard])
async def protected_route() -> dict:
    return {"message": "You're authenticated!"}

SQLAlchemy 集成

from litestar.contrib.sqlalchemy.plugins import SQLAlchemyPlugin
from litestar.contrib.sqlalchemy.config import SQLAlchemyAsyncConfig

db_config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///app.db",
)

sqlalchemy_plugin = SQLAlchemyPlugin(config=db_config)

app = Litestar(
    route_handlers=[UserController],
    plugins=[sqlalchemy_plugin],
)

中间件

from litestar.middleware import AbstractMiddleware
from litestar.types import ASGIApp, Receive, Scope, Send
import time

class TimingMiddleware(AbstractMiddleware):
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        start = time.time()
        await self.app(scope, receive, send)
        duration = time.time() - start
        print(f"Request took {duration:.3f}s")

app = Litestar(middleware=[TimingMiddleware])

WebSocket

from litestar import WebSocket, websocket

@websocket("/ws")
async def ws_handler(socket: WebSocket) -> None:
    await socket.accept()
    while True:
        data = await socket.receive_data()
        await socket.send_data(f"Echo: {data}")

常见问题

Q: 与 FastAPI 相比?

  • Litestar:更全面(DTO、Controller)、严格类型、性能相当
  • FastAPI:社区更大、教程更多、生态更成熟

Q: 性能如何?

与 FastAPI 相当,某些场景下更快(得益于编译优化和路由算法)。

Q: 文档自动生成?

内置 OpenAPI 文档,访问 /schema/swagger(Swagger UI)或 /schema/redoc(ReDoc)。


参考资源

  • GitHub:https://github.com/litestar-org/litestar
  • 文档:https://docs.litestar.dev/
  • 教程:https://docs.litestar.dev/latest/tutorials/
  • Discord:https://discord.gg/litestar