python-backend

python-backend

FastAPI、SQLAlchemy、Upstash 向けの本番環境対応 Python バックエンドのパターンです。

このスキルを使用する場面

• FastAPI で REST API を構築する
• JWT/OAuth2 認証を実装する
• SQLAlchemy の非同期データベースを設定する
• Redis/Upstash のキャッシュとレート制限を統合する
• AI が生成した Python コードをリファクタリングする
• API のパターンとプロジェクト構成を設計する

基本原則

1. 非同期を優先 - I/O 操作には async/await を使用します
2. すべてに型を指定 - 検証には Pydantic モデルを使用します
3. 依存性注入 - FastAPI の Depends() を使用します
4. 早期に失敗させる - 早い段階で検証し、HTTPException を使用します
5. デフォルトでセキュアに - ユーザー入力を決して信頼しません

クイックパターン

プロジェクト構成

src/
├── auth/
│   ├── router.py      # endpoints
│   ├── schemas.py     # pydantic models
│   ├── models.py      # db models
│   ├── service.py     # business logic
│   └── dependencies.py
├── posts/
│   └── ...
├── config.py
├── database.py
└── main.py

非同期ルート

# BAD - blocks event loop
@router.get("/")
async def bad():
    time.sleep(10)  # Blocking!

# GOOD - runs in threadpool
@router.get("/")
def good():
    time.sleep(10)  # OK in sync function

# BEST - non-blocking
@router.get("/")
async def best():
    await asyncio.sleep(10)  # Non-blocking

Pydantic による検証

from pydantic import BaseModel, EmailStr, Field

class UserCreate(BaseModel):
    email: EmailStr
    username: str = Field(min_length=3, max_length=50, pattern="^[a-zA-Z0-9_]+$")
    age: int = Field(ge=18)

依存性注入

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    payload = decode_token(token)
    user = await get_user(payload["sub"])
    if not user:
        raise HTTPException(401, "User not found")
    return user

@router.get("/me")
async def get_me(user: User = Depends(get_current_user)):
    return user

SQLAlchemy の非同期処理

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session

Redis キャッシュ

from upstash_redis import Redis

redis = Redis.from_env()

@app.get("/data/{id}")
def get_data(id: str):
    cached = redis.get(f"data:{id}")
    if cached:
        return cached
    data = fetch_from_db(id)
    redis.setex(f"data:{id}", 600, data)
    return data

レート制限

from upstash_ratelimit import Ratelimit, SlidingWindow

ratelimit = Ratelimit(
    redis=Redis.from_env(),
    limiter=SlidingWindow(max_requests=10, window=60),
)

@app.get("/api/resource")
def protected(request: Request):
    result = ratelimit.limit(request.client.host)
    if not result.allowed:
        raise HTTPException(429, "Rate limit exceeded")
    return {"data": "..."}

参考ドキュメント

詳しいパターンについては、以下を参照してください。

ドキュメント	内容
references/fastapi_patterns.md	プロジェクト構成、非同期処理、Pydantic、依存関係、テスト
references/security_patterns.md	JWT、OAuth2、パスワードのハッシュ化、CORS、API キー
references/database_patterns.md	SQLAlchemy の非同期処理、トランザクション、先行読み込み、マイグレーション
references/upstash_patterns.md	Redis、レート制限、QStash のバックグラウンドジョブ

リソース

• FastAPI ドキュメント
• SQLAlchemy 2.0 ドキュメント
• Upstash ドキュメント
• Pydantic ドキュメント