python-backend

Python バックエンド

asyncio、FastAPI、SQLAlchemy 2.0、接続プーリングを使って、本番環境向けの Python バックエンドを構築するためのパターンです。各カテゴリのルールは個別のファイルに記載され、必要に応じて rules/ から読み込まれます。

クイックリファレンス

カテゴリ	ルール	影響度	使用する場面
Asyncio	3	高	TaskGroup、構造化並行処理、キャンセル処理
FastAPI	3	高	依存関係、ミドルウェア、バックグラウンドタスク
SQLAlchemy	3	高	非同期セッション、リレーションシップ、マイグレーション
Pooling	3	中	データベースプール、HTTP セッション、チューニング

合計：4 つのカテゴリにわたる 12 個のルールです。 削られたファイルから復元されたプロジェクト内の決定事項は references/ork-delta.md に記載されています。ベンダーの資料は再掲せず、リンクしています（アップストリームのカバー範囲（再掲しない）を参照）。

クイックスタート

# FastAPI + SQLAlchemy async session
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# Reusable dependency alias (FastAPI's recommended Annotated form)
SessionDep = Annotated[AsyncSession, Depends(get_db)]

@router.get("/users/{user_id}")
async def get_user(user_id: UUID, db: SessionDep):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# Asyncio TaskGroup with timeout
async def fetch_all(urls: list[str]) -> list[dict]:
    async with asyncio.timeout(30):
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_url(url)) for url in urls]
    return [t.result() for t in tasks]

Asyncio

構造化並行処理、TaskGroup、Python 3.11 以降の機能を活用する最新の Python asyncio パターンです。

主なパターン

• TaskGroup は、構造化並行処理と自動キャンセルを備え、gather() に代わるものです
• asyncio.timeout() コンテキストマネージャーによる、組み合わせ可能なタイムアウト
• Semaphore による並行処理数の制限（HTTP リクエストのレート制限）
• 複数のタスク失敗を処理する、ExceptionGroup と except*
• 同期コードと非同期コードを橋渡しする asyncio.to_thread()

主な決定事項

決定事項	推奨事項
タスクの起動	gather() ではなく TaskGroup
タイムアウト	asyncio.timeout() コンテキストマネージャー
並行処理数の制限	asyncio.Semaphore
同期コードとの橋渡し	asyncio.to_thread()
キャンセル	CancelledError は必ず再送出する

FastAPI

ライフスパン、依存関係、ミドルウェア、設定に関する、本番環境向けの FastAPI パターンです。

主なパターン

• 起動・終了時のリソース管理に asynccontextmanager を使う ライフスパン
• クラスベースのサービスと Depends() による 依存性注入
• ミドルウェアスタック：CORS -> RequestID -> Timing -> Logging
• .env とフィールド検証を使う Pydantic Settings
• RFC 9457 Problem Details の本文を返すように設定する 例外ハンドラー（本文形式そのものは ork:api-design にあります）

主な決定事項

決定事項	推奨事項
ライフスパン	asynccontextmanager（イベントではなく）
依存関係	DI を使ったクラスベースのサービス
設定	.env を使った Pydantic Settings
レスポンス	パフォーマンス向上のため ORJSONResponse
ヘルスチェック	重要な依存関係をすべて確認する

SQLAlchemy

SQLAlchemy 2.0、AsyncSession、FastAPI との統合を使った非同期データベースのパターンです。

主なパターン

• expire_on_commit=False を指定し、リクエストごとに 1 つの AsyncSession を使う
• リレーションシップに lazy="raise" を設定し、意図しない N+1 クエリを防ぐ
• コレクションの先読み込みには selectinload を使う
• 汎用の非同期 CRUD を備えた リポジトリパターン
• メモリ管理のため、一括挿入を 1000～10000 行単位に分割する

主な決定事項

決定事項	推奨事項
セッションのスコープ	リクエストごとに 1 つの AsyncSession
遅延読み込み	lazy="raise" ＋明示的な読み込み
先読み込み	コレクションには selectinload
expire_on_commit	False（遅延読み込みエラーを防止）
プール	pool_pre_ping=True

Pooling

高性能な非同期 Python アプリケーション向けの、データベースと HTTP の接続プーリングです。

主なパターン

• pool_size、max_overflow、pool_pre_ping を使う SQLAlchemy プール
• min_size／max_size と接続のライフサイクルを指定する 直接利用の asyncpg プール
• TCPConnector の上限と DNS キャッシュを使う aiohttp セッション
• 起動時にプールを作成し、終了時に閉じる FastAPI のライフスパン
• Prometheus メトリクスによる プールの監視

プールサイズの計算式

pool_size = (concurrent_requests / avg_queries_per_request) * 1.5

この計算式で算出するのは、1 プロセス分のサイズです。サーバーの max_connections に対するフリート全体の上限と、プールのアラートしきい値は references/ork-delta.md に記載されています。

アンチパターン（禁止）

# NEVER use gather() for new code - no structured concurrency
# NEVER swallow CancelledError - breaks TaskGroup and timeout
# NEVER block the event loop with sync calls (time.sleep, requests.get)
# NEVER use global mutable state for db sessions
# NEVER skip dependency injection (create sessions in routes)
# NEVER share AsyncSession across tasks (race condition)
# NEVER use sync Session in async code (blocks event loop)
# NEVER create engine/pool per request
# NEVER forget to close pools on shutdown

アップストリームのカバー範囲（再掲しない）

2026-07-31 の wrap-plus-delta 間引きで削除されたトピックです。一次情報を参照してください。このスキルに含めるのは、ork のデルタ（社内ポリシー、既知の問題、稼働中の設定）だけです。ある行で社内サブセットを rules/ ファイルに残すと記載されている場合、そのファイルが OrchestKit の方針に関する正式な情報源であり、リンクはその周辺のベンダー側の情報だけを対象とします。

トピック	一次情報源
asyncio タスク API リファレンス：TaskGroup と gather() の比較、asyncio.timeout()、ExceptionGroup と except*、to_thread()。プロジェクト内で扱う範囲は rules/asyncio-taskgroup.md と rules/asyncio-cancellation.md に記載されています。	https://docs.python.org/3/library/asyncio-task.html
asyncio.Semaphore、Lock、Event、Queue のセマンティクス。プロジェクト内で扱う範囲は rules/asyncio-structured.md に記載されています。「一度だけ作成してタイムアウトを設定する」ルールは references/ork-delta.md に記載されています。	https://docs.python.org/3/library/asyncio-sync.html
イベントループのデバッグモード、slow_callback_duration、ブロッキング呼び出しの検出（プロジェクト内で定める 100 ms のしきい値は references/ork-delta.md に記載されています）	https://docs.python.org/3/library/asyncio-dev.html
FastAPI プロジェクトのひな形作成：アプリのレイアウト、APIRouter の構成、Pydantic Settings の組み込み、uvicorn のエントリーポイント	https://fastapi.tiangolo.com/tutorial/bigger-applications/
FastAPI のライフスパン API と起動／シャットダウンの仕組み。プロジェクト内で扱う範囲は rules/fastapi-background.md に記載されています。逆順でティアダウンするルールは references/ork-delta.md に記載されています。	https://fastapi.tiangolo.com/advanced/events/
Starlette／FastAPI のミドルウェア API、BaseHTTPMiddleware、call_next、CORS オプション。プロジェクト内で扱う範囲は rules/fastapi-middleware.md と references/fastapi-app-boilerplate.md に記載されています。ミドルウェアと依存関係の使い分けは references/ork-delta.md に記載されています。	https://fastapi.tiangolo.com/tutorial/middleware/
SQLAlchemy 2.0 の asyncio API：create_async_engine、async_sessionmaker、expire_on_commit、Mapped／mapped_column、with_for_update、一括挿入と更新。プロジェクト内で扱う範囲は rules/sqlalchemy-sessions.md、rules/sqlalchemy-relationships.md、rules/sqlalchemy-migrations.md、references/eager-loading.md に記載されています。	https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
SQLAlchemy のプール実装とオプション（pool_size、max_overflow、pool_pre_ping、pool_recycle、pool_timeout）。プロジェクト内で扱う範囲は rules/pooling-database.md に記載されています。	https://docs.sqlalchemy.org/en/20/core/pooling.html
計測用の SQLAlchemy プールイベント（checkout、checkin、connect）。プロジェクト内で扱う範囲は rules/pooling-tuning.md に記載されています。アラートのしきい値は references/ork-delta.md に記載されています。	https://docs.sqlalchemy.org/en/20/core/events.html
asyncpg プール API：create_pool、min_size／max_size、max_inactive_connection_lifetime、setup フック、型コーデック	https://magicstack.github.io/asyncpg/current/api/index.html
aiohttp クライアントリファレンス：ClientSession、TCPConnector の制限、キープアライブ、DNS キャッシュ、ClientTimeout。プロジェクト内で扱う範囲は rules/pooling-http.md に記載されています。	https://docs.aiohttp.org/en/stable/client_reference.html
PostgreSQL サーバー側の接続数制限（max_connections、superuser_reserved_connections）	https://www.postgresql.org/docs/current/runtime-config-connection.html
pytest-asyncio のフィクスチャと非同期テストのセットアップ	https://pytest-asyncio.readthedocs.io/en/stable/reference/fixtures/index.html
コンテナーおよび Kubernetes のパッケージ化、レディネス／ライブネスプローブ、リソース制限	https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
認証フロー、パスワードハッシュ化、トークンの有効期限ポリシー、セキュリティヘッダー	ork:security-patterns（src/skills/security-patterns/）が担当します。FastAPI の Depends() 認証チェーンは rules/fastapi-dependencies.md に記載されています。
RFC 9457 の問題詳細ボディ形式、API のバージョン管理、SSE および WebSocket の通信契約	ork:api-design（src/skills/api-design/）が担当します。FastAPI の例外ハンドラーの組み込みは rules/fastapi-background.md と references/fastapi-app-boilerplate.md に記載されています。
FastAPI、asyncio、SQLAlchemy、プーリングの本番環境向けチェックリスト	上記の情報源から導出できます。同じ内容のチェックリストは再掲していません。

関連スキル

• ork:architecture-patterns - クリーンアーキテクチャとレイヤー分離
• ork:async-jobs - バックグラウンド処理向けの Celery/ARQ
• ork:api-design - ワイヤーコントラクト、RFC 9457 エラー、SSE/WebSocket ストリーミング
• ork:database-patterns - データベーススキーマ設計
• ork:security-patterns - 認証、パスワードハッシュ化、トークンポリシー
• ork:testing-integration - pytest-asyncio と httpx ASGI のテスト環境設定
• ork:devops-deployment - Docker と Kubernetes のパッケージ化