Engineering FastAPIPydanticAsync Python

FastAPI Developer

FastAPI development: Pydantic v2 models, dependency injection, async patterns, background tasks, WebSocket endpoints, OpenAPI customization, and deployment.

A FastAPI expert covering Pydantic v2's model validators and computed fields, FastAPI's dependency injection for database sessions and auth, background task patterns with asyncio and Celery, and production deployment on gunicorn+uvicorn. Designs type-safe APIs where the OpenAPI schema and the Python types are always in sync.

Added Mar 20, 2026

$npx skills add johnefemer/skillfish --skill fastapi-developer

Capabilities Real Examples Get Started Models & Context Works Great With

What This Skill Can Do

Concrete capabilities you get when you install this skill.

Design Pydantic v2 models with field validators, model validators, and computed fields

Implement dependency injection for DB sessions, auth tokens, and feature flags

Build async endpoints with asyncpg or SQLAlchemy async and proper connection pooling

Add background tasks with FastAPI BackgroundTasks or Celery for long-running jobs

Implement WebSocket endpoints with connection management and pub/sub patterns

Deploy FastAPI with gunicorn + uvicorn workers, health checks, and Prometheus metrics

Real Conversations

See how the FastAPI Developer skill responds to actual requests. These are representative examples of what your AI agent will produce.

You

I want to implement JWT bearer token authentication in FastAPI using a dependency so I can protect routes with just Depends(get_current_user). How do I set this up?

Agent

Here's a complete JWT auth dependency using python-jose:

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from pydantic import BaseModel
import os

JWT_SECRET = os.environ["JWT_SECRET"]
ALGORITHM = "HS256"

bearer_scheme = HTTPBearer()

class TokenPayload(BaseModel):
    sub: str
    exp: int

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    token = credentials.credentials
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=[ALGORITHM])
        data = TokenPayload(**payload)
    except (JWTError, ValueError):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    user = await db.get(User, data.sub)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# Route usage — inject the dependency
@router.get("/me", response_model=UserResponse)
async def get_me(current_user: User = Depends(get_current_user)):
    return current_user

# Admin route — compose dependencies
async def require_admin(user: User = Depends(get_current_user)) -> User:
    if not user.is_admin:
        raise HTTPException(status_code=403, detail="Admin required")
    return user

The HTTPBearer scheme auto-populates the OpenAPI docs with a Bearer token input. Compose require_admin on top of get_current_user without duplicating auth logic.

You

My FastAPI endpoint runs fine for single requests (50ms) but degrades to 5 seconds under 100 concurrent requests. The endpoint does a single PostgreSQL query. What's happening?

Agent

50ms → 5s under 100 concurrent requests is a connection pool exhaustion pattern. Here's the diagnosis:

1. Check connection pool size:

# With asyncpg directly:
pool = await asyncpg.create_pool(dsn, min_size=5, max_size=10)
# With 10 max connections and 100 concurrent requests,
# 90 requests queue waiting for a connection slot

# Fix: increase max_size (or add a semaphore to reject excess early)
pool = await asyncpg.create_pool(dsn, min_size=10, max_size=50)

2. Verify you're not creating a new pool per request:

# ❌ New pool per request = new connections on every request (slow + exhausts PG max_connections)
@app.get("/data")
async def get_data():
    pool = await asyncpg.create_pool(dsn)  # BAD

# ✅ Pool created once in lifespan and stored on app.state
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await asyncpg.create_pool(dsn, min_size=5, max_size=50)
    yield
    await app.state.pool.close()

3. Check for accidental synchronous database calls: If you're using SQLAlchemy Core with a sync engine inside an async endpoint, it blocks the event loop thread. Use asyncpg or sqlalchemy.ext.asyncio with AsyncSession.

4. Monitor with query timing:

import time
async with pool.acquire() as conn:
    t0 = time.perf_counter()
    result = await conn.fetch(query)
    print(f"query: {time.perf_counter()-t0:.3f}s, pool_size: {pool.get_size()}")

The log will show whether time is spent waiting to acquire a connection vs. executing the query.

Get Started

How to install and use this skill in your preferred environment.

Skills are designed for AI coding agents (Claude Code, Cursor, Windsurf) and IDE-based workflows where the agent can read files, run scripts, and act on your codebase.

Models & Context

Which AI models and context windows work best with this skill.

Recommended Models

Claude Sonnet 4 or GPT-4o for FastAPI patterns and Pydantic v2 validators. Pydantic v2 has breaking changes from v1 — verify model syntax against the v2 migration guide.

Context Window

SKILL.md ~4KB. Fits in 32K context; large multi-router app structures benefit from 64K+.

Pro tips for best results

1

Be specific

Include numbers — users, budget, RPS — so the skill can size the architecture.

2

Share constraints

Compliance needs, team size, and existing stack all improve the output.

3

Iterate

Start with a high-level design, then ask follow-ups for IaC, cost analysis, or security review.

4

Combine skills

Pair with companion skills below for end-to-end coverage.

Works Great With

These skills complement FastAPI Developer for end-to-end coverage. Install them together for better results.

Python Pro

Advanced Python patterns: async/await, metaclasses, decorators, type hints, packaging, virtual environments, and performance optimization.

PythonAsync

Senior Backend

REST APIs, database optimization, authentication, and microservices.

BackendAPIs

Database Designer

Schema analysis, ERD generation, index optimization, and migrations.

DatabaseSchema

Docker Expert

Docker mastery: multi-stage builds, layer optimization, security scanning, Compose for dev/prod, networking, volumes, and container debugging.

DockerContainers

API Design Reviewer

REST API linter, breaking change detector, and design scorecard.

APIReview

$ skillfish add johnefemer/skillfish --all # install all skills at once

Ready to try FastAPI Developer?

Install the skill and start getting expert-level guidance in your workflow — any agent, any IDE.