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）、DI（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）にマッピングしてください。OpenTelemetry、Prometheus、Structlog、テンプレートプラグインは litestar.plugins.* からインポートし、非推奨の 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
例外と Problem Details	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
チャネルと Server-Sent Events	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／LLM エージェントの 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
wheel、PyApp スタンドアロンバイナリー、GitHub リリースマトリックス（litestar-build）	litestar-build
コンテナー、Cloud Run、GKE、Kubernetes、systemd、ランタイムサーバー（litestar-deployment）	litestar-deployment
Python／TypeScript／Litestar のコードスタイル、リンティング、型付け、品質ゲート	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. 機能全体でスタックの一貫性を保ちます。リポジトリですでに意図的に連携させている場合を除き、2 つの永続化レイヤー、2 つの DI コンテナー、または 2 つの設定方式を混在させないでください。

ガードレール

• 例から新しいスタックを推測しないでください。まず、現在のプロジェクトのデータレイヤー、DI パターン、設定スタイル、認証方式に合わせてください。
• 認証、認可、SQL クエリ、環境変数の解析、負荷の高いバックグラウンド処理をルートハンドラー内に直接記述しないでください。Guard の呼び出し可能オブジェクト、サービス／リポジトリ、キャッシュされた設定ファクトリー、タスクキューに委譲してください。
• 非推奨の StaticFilesConfig、litestar.contrib.* プラグイン、litestar.repository、litestar.params.Dependency、暗黙的な依存関係パラメーターを使用しないでください。
• 一括代入や、隠れたリレーションシップの遅延読み込みを防ぐ exclude／include フィールドルールを設けずに、ORM モデルを dto／return_dto で直接公開しないでください。
• リクエストハンドラー内、または設定モジュール外のモジュールレベルコードで 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