Uvicornは、Pythonエコシステムにおいて広く採用されている高性能なASGI（Asynchronous Server Gateway Interface）サーバー実装です。標準的なasyncioイベントループを基盤とし、Cythonによる最適化を組み合わせており、大規模な同時接続や低レイテンシが求められるバックエンドサービス向けに設計されています。

概要と動作原理

従来のWSGI方式がスレッドやプロセスベースで同期的にリクエストを処理するのに対し、UvicornはノンブロッキングI/Oとコルーチンによって複数の接続を単一スレッド内で効率的に切り替えます。これにより、FastAPI、Starlette、Django 4+などの現代的なPythonフレームワークとシームレスに連携し、高いスループットを実現します。

導入準備

パッケージマネージャーを使用して依存関係を解決します。[standard]フラグには、開発時の自動リロードツールやSSL関連モジュールが含まれるため、本番・開発両方で推奨されます。

$ pip install uvicorn[standard]

クイックスタート

基本的なルーティング構成と起動コマンドの例です。変数名やエンドポイント構造を調整し、実際のプロジェクトに近い形式に整えています。

# main_service.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="UvicornDemo", version="1.0.0")

class SystemStatus(BaseModel):
    phase: str
    uptime_sec: float

@app.get("/v1/status")
async def retrieve_health():
    return SystemStatus(phase="stable", uptime_sec=3600.5)

$ uvicorn main_service:app --host 0.0.0.0 --port 8080 --log-level info

主要設定パラメータ

• --host: リッスン対象のネットワークインターフェース（デフォルト: 127.0.0.1）
• --port: TCPポート番号（デフォルト: 8000）
• --workers: ワーカープロセス数。複数指定することでマルチコアCPUの性能を引き出せます。
• --loop: イベントループ実装の選択。auto, asyncio, uvloopから指定可能です。
• --ws: WebSocketプロトコルエンジン。websocketsまたはwsprotoを選択できます。

高度な機能実装パターン

TLS/SSL暗号化通信

プロダクション環境では必須となるHTTPS対応です。コマンドライン引数で証明書パスを渡すことで、内部でのTLSハンドシェイクを有効化できます。

$ uvicorn main_service:app --ssl-keyfile ./cert/server.key --ssl-certfile ./cert/server.crt

双方向WebSocketストリーミング

リアルタイムデータ配信用に、接続管理ロジックを強化したサンプルです。切断検知とクライアントリストのクリーンアップを実装しています。

# ws_feed.py
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

feed_app = FastAPI()
active_sockets: list[WebSocket] = []

@feed_app.websocket("/stream/events")
async def event_stream(socket: WebSocket):
    await socket.accept()
    active_sockets.append(socket)
    try:
        while True:
            payload = await socket.receive_text()
            dead_peers = []
            for peer in active_sockets:
                if peer != socket:
                    try:
                        await peer.send_text(f"echo:{payload}")
                    except RuntimeError:
                        dead_peers.append(peer)
            for dp in dead_peers:
                active_sockets.remove(dp)
    except WebSocketDisconnect:
        pass
    finally:
        if socket in active_sockets:
            active_sockets.remove(socket)

リクエスト前処理ミドルウェア

信頼できるプロキシからのX-Forwardedヘッダー処理や、CORSポリシーの適用例です。

# mw_pipeline.py
from fastapi import FastAPI
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
from starlette.middleware.cors import CORSMiddleware

pipeline_app = FastAPI()

pipeline_app.add_middleware(
    ProxyHeadersMiddleware,
    trusted_hosts=["10.0.0.0/8", "192.168.1.0/24"]
)
pipeline_app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization"]
)

@pipeline_app.get("/data/payload")
async def get_payload():
    return {"value": 42}

バックグラウンドタスクスケジューリング

メインイベントループをブロックすることなく定期処理を実行するパターンです。BackgroundTasksを用いた非同期実行例を確認します。

# bg_scheduler.py
from fastapi import FastAPI, BackgroundTasks
import asyncio

sched_app = FastAPI()

async def periodic_worker(interval_sec: float, label: str):
    cycles = 0
    while True:
        cycles += 1
        await asyncio.sleep(interval_sec)
        print(f"[{label}] Completed cycle {cycles}")

@sched_app.on_event("startup")
def register_workers(background_tasks: BackgroundTasks):
    background_tasks.add_task(periodic_worker, 5.0, "metrics_collector")
    background_tasks.add_task(periodic_worker, 15.0, "cache_flusher")

@sched_app.post("/trigger/manual")
async def manual_trigger():
    return {"status": "scheduled"}

カスタム例外ハンドリング

フレームワーク既定のエラーフォーマットを上書きし、構造化されたJSONレスポンスを返却する実装です。

# custom_errors.py
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from pydantic import ValidationError

err_app = FastAPI()

@err_app.exception_handler(ValidationError)
async def handle_validation(req: Request, exc: ValidationError):
    return JSONResponse(
        status_code=422,
        content={"code": "VALIDATION_ERROR", "reason": str(exc)}
    )

@err_app.get("/items/{item_idx}")
async def lookup_item(item_idx: int):
    if item_idx not in range(1, 101):
        raise HTTPException(status_code=404, detail="Item out of bounds")
    return {"idx": item_idx, "type": "reference"}

実践的ユースケース

コンカレントAPIゲートウェイ

I/O待機が長時間発生する外部サービス呼び出しを非同期で並列実行し、リソース効率を最大化する構成です。

# concurrent_gw.py
from fastapi import FastAPI
import aiohttp

gw_app = FastAPI()

async def call_remote(target_url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(target_url) as res:
            res.raise_for_status()
            return await res.json()

@gw_app.get("/proxy/fetch")
async def bridge_service(url: str):
    result = await call_remote(url)
    return result

イベント駆動型WebSocketブローカー

複数のクライアント間でメッセージを中継し、ステートフルな接続維持を行うための実装例です。前述のws_feed.pyパターンを拡張し、認証トークンの簡易検証を含む構成にすることも可能です。