litestar

Litestar 核心框架与生态系统中心

在使用所有 Litestar 框架核心子系统（应用组合、路由、控制器、参数、DTO、OpenAPI、依赖注入、Dishka 集成、身份验证/守卫、中间件、异常处理、插件协议、WebSocket、Channels/SSE、类型化设置，以及 CRUD 数据服务/分页）时，请使用此技能；也可借此转到独立的 Litestar 生态技能。

本指南针对已发布的 Litestar 2.25.0 契约。请勿在使用者示例中照搬尚未发布的 main 分支 API。

代码风格规则

• 优先使用 Litestar 第一方生态软件包（litestar-granian、advanced-alchemy、sqlspec、msgspec、litestar-saq、litestar-queues、litestar-vite、litestar-autowire、litestar-security、litestar-email、litestar-mcp、litestar-htmx），并与项目现有技术栈保持一致，涵盖数据访问（advanced-alchemy 与 sqlspec）、依赖注入（Provide 与 Dishka）、设置（dataclass 与 pydantic-settings）、后台任务（litestar-saq 与 litestar-queues）和 Channels 后端（Memory、Redis、AsyncPg/PsycoPg、SQLSpec）。
• 保持 Litestar(...) 应用设置和路由处理程序精简：将领域路由组织到 Controller / Router 层级中（urls.py + controllers/ + services.py + schemas.py + dependencies.py + guards.py），并在 ApplicationCore(InitPluginProtocol, CLIPluginProtocol) + create_app() -> Litestar 中集中处理跨领域的装配（litestar-fullstack / litestar-sqlstack 模式）。
• 使用 Litestar 2.22–2.25 的显式注解：FromPath[T]、FromQuery[T]、FromHeader[T]、FromCookie[T]（需要约束或线上名称时，使用 PathParameter、QueryParameter、HeaderParameter、CookieParameter）、JSONBody[T]、RequestBody[T]、MsgPackBody[T]、MultipartBody[T]、URLEncodedBody[T]、RawBody、QueryParams、RequestCookies、NamedDependency[T] 和 SkipValidation[T]。不要引入隐式依赖、未指定类型的保留关键字参数（如没有请求体标记的 data: MyDTO 或 body: bytes），也不要使用已弃用的 litestar.params.Dependency。
• 定义显式的 msgspec.Struct（使用 rename="camel" 设置 camelCase 线上格式）或 MsgspecDTO / SQLAlchemyDTO / DataclassDTO 边界契约（在路由和控制器中使用 dto 和 return_dto）；切勿在没有写入安全 DTO 的情况下直接暴露原始 ORM 模型。
• 通过 AbstractAuthenticationMiddleware 或第一方 JWTAuth / JWTCookieAuth / SessionAuth 强制执行身份验证，并在 Guard 函数 (connection: ASGIConnection, route_handler: BaseRouteHandler) -> None | Awaitable[None] 或 litestar-security 装饰器中强制执行授权——绝不要在路由处理程序的函数体中执行。
• 优先使用内置中间件配置（CORSConfig、CSRFConfig、AllowedHostsConfig、CompressionConfig、RateLimitConfig、ResponseCacheConfig、LoggingMiddlewareConfig），而非自定义中间件；对于自定义横切关注点，请继承 ASGIMiddleware（litestar.middleware.ASGIMiddleware），并使用 scopes 和 exclude 过滤器。
• 在服务中引发特定于领域的 ApplicationError 子类，并在 exception_handlers 中将其映射为 HTTP 响应（或 RFC 9457 Problem Details）；请从 litestar.plugins.* 导入 OpenTelemetry、Prometheus、Structlog 和模板插件（绝不要使用已弃用的 litestar.contrib.*）。
• 所有请求处理程序、WebSocket/SSE 流、服务/仓储方法、存储和外部客户端均使用异步 I/O。

快速参考

核心框架子系统（此技能）

子系统	关键符号／模式	参考
应用组合与生命周期	Litestar(...)、AppConfig、ApplicationCore、lifespan、State、ImmutableState、2.22–2.25 契约	app-and-lifecycle.md
CLI 与项目配置	自动发现、LITESTAR_* 环境变量、litestar run/info/routes/schema/sessions、CLIPlugin、litestar.toml	cli-and-config.md
路由处理器与控制器	@get/@post/@put/@patch/@delete、Controller、Router、分层配置继承	handlers.md
参数与请求体	FromPath、FromQuery、FromHeader、FromCookie、JSONBody、RequestBody、MsgPackBody、MultipartBody、URLEncodedBody、RawBody、QueryParams、RequestCookies、UploadFile	parameters.md
领域包布局	domains/<name>/ 结构、子控制器、全栈领域接线	layout.md
DTO 与边界契约	MsgspecDTO、SQLAlchemyDTO、DataclassDTO、PydanticDTO、DTOConfig、DTOData[T]、DTOField、Mark	dtos.md
OpenAPI 与架构 UI	OpenAPIConfig、Scalar/Swagger/Redoc/Stoplight/RapiDoc 插件、ResponseSpec、Operation、安全方案	openapi.md
依赖注入与 Dishka	Provide、NamedDependency[T]、SkipValidation[T]、生成器清理、FromDishka[T]、setup_dishka	di-and-dishka.md
身份验证与守卫	Guard 可调用对象、ASGIConnection、JWTAuth、JWTCookieAuth、SessionAuth、OAuth2、RBAC 与租户守卫	auth-and-guards.md
中间件与 HTTP 安全	ASGIMiddleware、DefineMiddleware、CORSConfig、CSRFConfig、AllowedHostsConfig、CompressionConfig、RateLimitConfig	middleware.md
异常与问题详情	HTTPException 层次结构、ApplicationError 映射、exception_handlers、after_exception、RFC 9457	exceptions.md
插件协议	InitPlugin、CLIPluginProtocol、SerializationPluginProtocol、OpenAPISchemaPluginProtocol、DIPlugin、PydanticPlugin	plugins.md
WebSocket	@websocket、WebSocket 生命周期、@websocket_listener、@websocket_stream、WS Dishka DI、WS 守卫、stream_pubsub	websockets.md
通道与服务器发送事件	ChannelsPlugin（Memory、Redis、AsyncPg、PsycoPg、SQLSpec）、ServerSentEvent、Stream、RealtimeEvent、RealtimePublisher	channels-and-sse.md
类型化设置与环境配置	dataclass + get_env 与 pydantic-settings、@lru_cache 工厂、插件配置构建器	settings.md
筛选器与分页	create_filter_dependencies、FilterTypes、OffsetPagination、ClassicPagination、游标分页、自定义搜索	filters-and-pagination.md
数据服务与仓储	SQLAlchemyAsyncRepositoryService（advanced-alchemy）和 SQLSpecAsyncService（sqlspec）CRUD 服务	services-and-repos.md
日志记录与可观测性	LoggingConfig、StructlogPlugin、OpenTelemetryPlugin、PrometheusConfig、PrometheusController	logging-and-observability.md
存储与响应缓存	StoreRegistry、MemoryStore、FileStore、RedisStore、ValkeyStore、ResponseCacheConfig	stores-and-caching.md

独立生态技能

任务	技能
基于约定发现领域包（AutowirePlugin、domain_packages）	litestar-autowire
msgspec.Struct、Meta、标记联合、JSON/MsgPack 编码/解码钩子	msgspec
SecurityPlugin、CurrentUser、Principal、@requires_role/@requires_scope/@requires_tenant	litestar-security
SQLAlchemy ORM 模型、SQLAlchemyPlugin、Alembic 迁移、仓库	advanced-alchemy
SQL 优先驱动、SQLFileLoader、查询构建器、ADK 存储、数据字典	sqlspec
Vite 资源流水线、HMR、TypeGen、SPA/混合模式，以及 Inertia.js（InertiaConfig、component=）	litestar-vite
服务端渲染的 HTMX 请求、局部模板、HX-* 响应辅助工具	litestar-htmx
基于 Redis 的后台工作进程、队列、cron 计划（litestar-saq）	litestar-saq
SQL/内存任务队列、@task、QueueService、工作进程（litestar-queues）	litestar-queues
事务性邮件（EmailPlugin、SMTP/Resend/SendGrid/Mailgun/SES 后端）	litestar-email
Model Context Protocol 服务（LitestarMCP、@mcp.tool/@mcp.resource/@mcp.prompt）	litestar-mcp
Google ADK / 大模型 Agent 的 HTTP 和 SSE 服务、工具调用、会话服务	litestar-ai
Granian ASGI 服务插件、运行时线程、HTTP/2、TLS、静态挂载	litestar-granian
TestClient、AsyncTestClient、create_async_test_client、处理器/DI/Guard 测试	litestar-testing
类型化测试数据工厂（MsgspecFactory、ModelFactory、DataclassFactory）	polyfactory
用于集成测试的 Docker 数据库/服务夹具（pytest-databases）	pytest-databases
Wheels、PyApp 独立二进制文件、GitHub 发布矩阵（litestar-build）	litestar-build
容器、Cloud Run、GKE、Kubernetes、systemd、运行时服务（litestar-deployment）	litestar-deployment
Python/TypeScript/Litestar 代码风格、Lint、类型检查和质量门禁	litestar-styleguide

## 工作流

1. 先检测项目技术栈：检查 pyproject.toml、litestar.toml、现有领域包、DI 设置（Provide 还是 Dishka）、数据层（advanced-alchemy 还是 sqlspec）、设置（dataclass 还是 pydantic-settings），以及身份验证（JWTAuth/SessionAuth 还是 litestar-security）。
2. 针对 Litestar 框架相关事项，打开匹配的核心参考文档，查看 references/*.md：
   • 路由、控制器、参数或领域布局 -> handlers.md、parameters.md、layout.md
   • DTO、请求/响应架构或 OpenAPI 文档 -> dtos.md、openapi.md
   • 依赖注入或 Dishka 提供者 -> di-and-dishka.md
   • 身份验证后端、守卫、中间件或异常处理器 -> auth-and-guards.md、middleware.md、exceptions.md
   • 插件、WebSocket、Channels/SSE、设置或 CRUD 服务/筛选器 -> plugins.md、websockets.md、channels-and-sse.md、settings.md、filters-and-pagination.md、services-and-repos.md
   • 应用组合、CLI、日志/指标/追踪或存储/缓存 -> app-and-lifecycle.md、cli-and-config.md、logging-and-observability.md、stores-and-caching.md
3. 处理专用的一方软件包时，打开独立的生态技能文档（advanced-alchemy、sqlspec、msgspec、litestar-vite、litestar-htmx、litestar-saq、litestar-queues、litestar-security、litestar-autowire、litestar-email、litestar-mcp、litestar-ai、litestar-granian、litestar-testing、polyfactory、pytest-databases、litestar-build、litestar-deployment）。
4. 在整个功能中保持技术栈一致：除非代码库已主动整合，否则不要混用两种持久化层、两种 DI 容器或两种设置范式。

防护规则

• 不要从示例中推断新的技术栈——始终先匹配当前项目的数据层、DI 模式、设置风格和身份验证机制。
• 不要将身份验证、授权、SQL 查询、环境变量解析或繁重的后台工作直接放在路由处理器中；应交由 Guard 可调用对象、服务/仓库、缓存的设置工厂和任务队列处理。
• 不要使用已弃用的 StaticFilesConfig、litestar.contrib.* 插件、litestar.repository、litestar.params.Dependency 或隐式依赖参数。
• 不要在 dto / return_dto 中直接暴露 ORM 模型，除非使用 exclude / include 字段规则来防止批量赋值和隐藏关系的延迟加载。
• 不要在请求处理器内或设置模块之外的模块级代码中调用 os.getenv()，也绝不要在源文件中硬编码密钥。
• 对于不支持历史记录的 Channels 后端（RedisChannelsPubSubBackend、AsyncPgChannelsBackend、PsycoPgChannelsBackend），不要调用 start_subscription(..., history=N)。

## 验证检查点

• 路由处理器使用 Litestar 2.25 的显式参数、请求体、保留关键字参数和依赖标记（FromPath[T]、FromQuery[T]、JSONBody[T]/RequestBody[T]、RawBody、QueryParams、RequestCookies、NamedDependency[T]、SkipValidation[T]）。
• 请求和响应契约使用 msgspec.Struct 或 MsgspecDTO / SQLAlchemyDTO，并显式声明字段排除项和 status_code。
• 授权检查应放在 Guard 函数（ASGIConnection、BaseRouteHandler）或 litestar-security 装饰器中；领域异常应通过 exception_handlers 进行清晰映射。
• 内置配置对象（CORSConfig、CSRFConfig、AllowedHostsConfig、CompressionConfig、LoggingConfig、OpenAPIConfig、ResponseCacheConfig）应通过 Litestar(...) 或 ApplicationCore.on_app_init 接入。
• 在此仓库中编辑技能时，验证命令应使用 make 目标。

示例

import click
import msgspec
from litestar import Controller, Litestar, MediaType, Request, Response, get, post
from litestar.config.app import AppConfig
from litestar.config.cors import CORSConfig
from litestar.config.response_cache import ResponseCacheConfig
from litestar.connection import ASGIConnection
from litestar.datastructures import State
from litestar.di import NamedDependency, Provide
from litestar.dto import DTOConfig, MsgspecDTO
from litestar.exceptions import LitestarException, NotFoundException, PermissionDeniedException
from litestar.handlers.base import BaseRouteHandler
from litestar.logging import LoggingConfig
from litestar.openapi import OpenAPIConfig
from litestar.openapi.plugins import ScalarRenderPlugin
from litestar.params import FromPath, FromQuery, JSONBody
from litestar.plugins import CLIPluginProtocol, InitPluginProtocol
from litestar.status_codes import HTTP_201_CREATED, HTTP_409_CONFLICT
from litestar.stores.memory import MemoryStore
from litestar.stores.registry import StoreRegistry


class DuplicateSkuError(LitestarException):
    """Raised when an item SKU already exists."""

    def __init__(self, sku: str) -> None:
        self.sku = sku
        self.message = f"SKU '{sku}' already exists"
        super().__init__(self.message)


def duplicate_sku_exception_handler(
    request: Request[object, object, State],
    exc: DuplicateSkuError,
) -> Response[dict[str, object]]:
    """Map DuplicateSkuError to an HTTP 409 response."""
    return Response(
        media_type=MediaType.JSON,
        status_code=HTTP_409_CONFLICT,
        content={"detail": exc.message, "sku": exc.sku, "path": request.url.path},
    )


async def require_catalog_editor(
    connection: ASGIConnection[BaseRouteHandler, dict[str, object], object, State],
    _: BaseRouteHandler,
) -> None:
    """Verify the authenticated user has catalog write permission."""
    user = connection.scope.get("user")
    if not isinstance(user, dict) or "catalog:write" not in user.get("scopes", ()):
        raise PermissionDeniedException("Missing catalog:write scope")


class CatalogItem(msgspec.Struct, rename="camel"):
    """Serialized catalog item contract."""

    id: int
    sku: str
    name: str
    locale: str = "en"


class CreateCatalogItem(msgspec.Struct, rename="camel"):
    """Payload for creating a catalog item."""

    sku: str
    name: str


class CatalogItemReadDTO(MsgspecDTO[CatalogItem]):
    """Read DTO for CatalogItem responses."""

    config = DTOConfig(rename_strategy="camel")


class CatalogService:
    """Domain service managing catalog items."""

    async def fetch_item(self, item_id: int, locale: str) -> CatalogItem:
        if item_id <= 0:
            raise NotFoundException(f"Item {item_id} not found")
        return CatalogItem(id=item_id, sku=f"SKU-{item_id}", name="Widget", locale=locale)

    async def create_item(self, data: CreateCatalogItem) -> CatalogItem:
        if data.sku == "DUPLICATE":
            raise DuplicateSkuError(data.sku)
        return CatalogItem(id=1, sku=data.sku, name=data.name)


async def provide_catalog_service() -> CatalogService:
    """Provide request-scoped CatalogService."""
    return CatalogService()


class CatalogController(Controller):
    """HTTP controller for catalog operations."""

    path = "/catalog"
    tags = ["Catalog"]
    dependencies = {"catalog_service": Provide(provide_catalog_service)}
    return_dto = CatalogItemReadDTO

    @get("/{item_id:int}", cache=60)
    async def get_item(
        self,
        item_id: FromPath[int],
        catalog_service: NamedDependency[CatalogService],
        locale: FromQuery[str] = "en",
    ) -> CatalogItem:
        """Resolve path/query parameters and service dependency explicitly."""
        return await catalog_service.fetch_item(item_id=item_id, locale=locale)

    @post("/", guards=[require_catalog_editor], status_code=HTTP_201_CREATED)
    async def create_item(
        self,
        data: JSONBody[CreateCatalogItem],
        catalog_service: NamedDependency[CatalogService],
    ) -> CatalogItem:
        """Validate JSON payload, enforce guard, and delegate creation to service."""
        return await catalog_service.create_item(data)


class ApplicationCore(InitPluginProtocol, CLIPluginProtocol):
    """Reference architecture core plugin (litestar-fullstack / litestar-sqlstack)."""

    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        app_config.route_handlers.append(CatalogController)
        app_config.exception_handlers[DuplicateSkuError] = duplicate_sku_exception_handler
        app_config.cors_config = CORSConfig(allow_origins=["https://app.example.com"])
        app_config.openapi_config = OpenAPIConfig(
            title="Catalog API",
            version="1.0.0",
            render_plugins=[ScalarRenderPlugin()],
        )
        app_config.stores = StoreRegistry(stores={"response_cache": MemoryStore()})
        app_config.response_cache_config = ResponseCacheConfig(default_expiration=60)
        app_config.logging_config = LoggingConfig(
            log_exceptions="always",
            disable_stack_trace={404, NotFoundException},
        )
        app_config.state = State({"service_name": "catalog-api"})
        return app_config

    def on_cli_init(self, cli: click.Group) -> None:
        _ = cli


def create_app() -> Litestar:
    """Compose the Litestar 2.25 application via ApplicationCore."""
    return Litestar(plugins=[ApplicationCore()])

参考索引

• app-and-lifecycle.md
• auth-and-guards.md
• channels-and-sse.md
• cli-and-config.md
• di-and-dishka.md
• dtos.md
• exceptions.md
• filters-and-pagination.md
• handlers.md
• layout.md
• logging-and-observability.md
• middleware.md
• openapi.md
• parameters.md
• plugins.md
• services-and-repos.md
• settings.md
• stores-and-caching.md
• websockets.md

官方参考资料

• https://docs.litestar.dev/ - Litestar 文档
• https://docs.litestar.dev/latest/reference/ - Litestar API 参考
• https://docs.litestar.dev/latest/release-notes/changelog.html - Litestar 2.x 更新日志
• https://github.com/litestar-org/litestar/tree/v2.25.0 - 已审查的 Litestar 2.25.0 源码
• https://dishka.readthedocs.io/en/stable/integrations/litestar.html - Dishka Litestar 集成

共享风格指南基线

• 通用
• Python
• Litestar