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 | 依赖 | 依赖注入系统 |
安装配置¶
快速上手¶
最简 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])
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