Expert in FastAPI async framework with Python type hints, Pydantic models, dependency injection, OAuth2, JWT auth, and production deployment. Use when you need deep expertise in fastapi.
Scanned 9/8/2026
Install to Claude Code
Skills Directorynpx -y skills add anubhavg-icpl/vibe --skill fastapi-expert --agent claude-codeInstalls into .claude/skills of the current project.
Are you the author of Fastapi Expert?
Add the live security badge to your README — it updates automatically with every re-scan.
[](https://www.skillsdirectory.com/skills/anubhavg-icpl-fastapi-expert)More formats (shields.io, HTML) on the badges page.
---
name: fastapi-expert
description: Expert in FastAPI async framework with Python type hints, Pydantic models, dependency injection, OAuth2, JWT auth, and production deployment. Use when you need deep expertise in fastapi.
license: CC-BY-NC-SA-4.0
metadata:
risk: unknown
source: community
kind: mode
category: backend
tags: [fastapi, python, async, backend, framework, api]
---
# FastAPI Expert Mode
## Overview
You are an expert FastAPI framework specialist with deep knowledge of async/await, Pydantic models, dependency injection, OAuth2, JWT authentication, background tasks, WebSocket, OpenAPI documentation, and production deployment.
## Core Principles
1. **Async First** - Leverage async/await for I/O operations
2. **Type Safety** - Full Python type hints throughout
3. **Pydantic Models** - Automatic validation, serialization
4. **Dependency Injection** - FastAPI's built-in dependency injection
5. **OAuth2 Support** - Authorization code flow with PKCE
6. **Background Tasks** - Offload heavy operations to worker threads
7. **OpenAPI** - Auto-generated API documentation
8. **Production Ready** - Gunicorn/UVicorn with proper configuration
## Basic Setup
### Application Structure
```
app/
├── main.py
├── models/
│ ├── __init__.py
│ ├── user.py
│ └── post.py
├── schemas/
│ ├── user.py
│ └── post.py
├── routers/
│ ├── auth.py
│ ├── users.py
│ └── posts.py
├── core/
│ ├── config.py
│ ├── security.py
│ └── database.py
├── main.py
├── requirements.txt
└── pyproject.toml
```
### Main Application
```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from core.config import settings
from routers import auth, users, posts
app = FastAPI(
title=settings.PROJECT_NAME,
description=settings.PROJECT_DESCRIPTION,
version=settings.VERSION,
docs_url="/docs",
redoc_url="/redoc",
)
# CORS middleware
app.add_middleware(
CORSMiddleware(
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
)
# Include routers
app.include_router(auth.router, prefix="/auth", tags=["Authentication"])
app.include_router(users.router, prefix="/users", tags=["Users"])
app.include_router(posts.router, prefix="/posts", tags=["Posts"])
@app.on_event("startup")
async def startup_event():
# Initialize database connection
from core.database import engine
await engine.connect()
@app.on_event("shutdown")
async def shutdown_event():
# Close database connection
from core.database import engine
await engine.disconnect()
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"main:app",
host=settings.HOST,
port=settings.PORT,
reload=settings.DEBUG,
)
```
## Pydantic Models
### Base Model
```python
# models/user.py
from pydantic import BaseModel, EmailStr, validator
from datetime import datetime
from typing import Optional
class UserBase(BaseModel):
email: EmailStr
first_name: Optional[str] = None
last_name: Optional[str] = None
is_active: bool = True
created_at: datetime = None
updated_at: datetime = None
@validator('email')
def email_must_be_lower(cls, v: str):
return v.lower()
class Config:
json_schema_extra = "forbid"
```
### Create Model
```python
# models/user.py
from pydantic import BaseModel
from datetime import datetime
class UserCreate(BaseModel):
email: EmailStr
password: str
first_name: Optional[str] = None
last_name: Optional[str] = None
role: str = "user"
class UserUpdate(BaseModel):
email: Optional[EmailStr] = None
first_name: Optional[str] = None
last_name: Optional[str] = None
role: Optional[str] = None
is_active: Optional[bool] = None
class UserInDB(UserBase):
id: int
created_at: datetime
updated_at: datetime
```
### Response Models
```python
# schemas/user.py
from pydantic import BaseModel
from models.user import UserInDB
class UserResponse(BaseModel):
id: int
email: str
first_name: Optional[str]
last_name: Optional[str]
is_active: bool
created_at: datetime
updated_at: datetime
class UserListResponse(BaseModel):
users: list[UserResponse]
total: int
page: int
per_page: int
```
## Dependency Injection
### FastAPI Depends
```python
# routers/users.py
from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy import select
from typing import List
from core.database import get_db
from models.user import User, UserInDB
from schemas.user import UserResponse
async def get_current_user(db: Session = Depends(get_db)) -> UserInDB | None:
# Get user from JWT token
user = db.execute(select(User).where(User.email == "user@example.com")).scalar()
return user
async def get_user_or_404(db: Session = Depends(get_db)) -> UserInDB:
user = await get_current_user(db)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
```
### Background Tasks
### Simple Background Task
```python
# core/tasks.py
from fastapi import BackgroundTasks
import time
import asyncio
background_tasks = BackgroundTasks()
@background_tasks.task
def send_welcome_email(email: str, username: str):
print(f"Sending welcome email to {email}...")
time.sleep(2) # Simulate email sending
print(f"Welcome email sent to {email} for {username}")
# Usage in routes
from fastapi import BackgroundTasks
from core.tasks import background_tasks
def trigger_welcome(user_email: str, username: str):
background_tasks.send_welcome_email(user_email, username)
```
### Advanced Background Task
```python
# core/tasks.py
from fastapi import BackgroundTasks
import asyncio
background_tasks = BackgroundTasks()
@background_tasks.task
def process_data(data: list):
results = []
for item in data:
# Simulate processing
result = await asyncio.sleep(1)
results.append(f"Processed: {item}")
return results
```
## OAuth2
### Authorization Code Flow
```python
# routers/auth.py
from fastapi import FastAPI, Depends, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.security.oauth2 import OAuth2PasswordRequestForm
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
class OAuth2PasswordRequestForm(OAuth2PasswordRequestForm):
grant_type: str = "password"
username: str
password: str
scope: str = "read write"
client_id: str = settings.OAUTH_CLIENT_ID
client_secret: str = settings.OAUTH_CLIENT_SECRET
redirect_uri: str = settings.OAUTH_REDIRECT_URI
@app.post("/token")
async def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends()
):
return await authenticate_user(form_data.username, form_data.password)
@app.get("/users/me")
async def get_users_me(current_user: str = Depends(oauth2_scheme)):
return {"username": current_user}
```
### JWT Authentication
### JWT Utils
```python
# core/security.py
from datetime import datetime, timedelta
from typing import Optional
import jose
from core.config import settings
def create_access_token(data: dict) -> str:
to_encode = data.copy()
to_encode.update({
"exp": datetime.utcnow() + timedelta(minutes=15),
"iat": datetime.utcnow(),
"type": "access",
})
return jose.encode(to_encode, settings.JWT_SECRET_KEY, algorithm="HS256")
def create_refresh_token(data: dict) -> str:
to_encode = data.copy()
to_encode.update({
"exp": datetime.utcnow() + timedelta(days=7),
"iat": datetime.utcnow(),
"type": "refresh",
})
return jose.encode(to_encode, settings.JWT_SECRET_KEY, algorithm="HS256")
def decode_token(token: str) -> Optional[dict]:
try:
payload = jose.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
return payload
except jose.JWTError:
return None
```
### JWT Protected Routes
```python
# routers/posts.py
from fastapi import Depends, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from core.security import decode_token
from typing import List
oauth2_scheme = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(oauth2_scheme)
) -> Optional[dict]:
payload = decode_token(credentials.credentials)
if not payload or payload.get("type") != "access":
raise HTTPException(status_code=401, detail="Invalid or expired token")
return payload
@app.get("/posts")
async def get_posts(
current_user: dict = Depends(get_current_user),
skip: int = 0,
limit: int = 10,
db: Session = Depends(get_db)
):
posts = db.execute(select(Post).order_by(Post.created_at.desc()).offset(skip).limit(limit))
return {"posts": posts}
```
## Testing
### Pytest Configuration
```python
# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from core.config import settings
# Test database
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
TestSessionLocal = scoped_session(sessionmaker(bind=engine, autoflush=False))
@pytest.fixture(scope="session")
def db():
yield TestSessionLocal()
TestSessionLocal.remove()
```
### Route Tests
```python
# tests/test_main.py
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_create_user():
response = client.post(
"/auth/register",
json={
"email": "test@example.com",
"password": "password123",
"first_name": "Test",
"last_name": "User",
}
)
assert response.status_code == 201
assert "email" in response.json()
assert "id" in response.json()
def test_login():
response = client.post(
"/auth/login",
json={
"username": "test@example.com",
"password": "password123",
}
)
assert response.status_code == 200
assert "access_token" in response.json()
def test_protected_route():
# First, register and login
user_response = client.post(
"/auth/register",
json={
"email": "test@example.com",
"password": "password123",
}
)
assert user_response.status_code == 201
login_response = client.post(
"/auth/login",
json={
"username": "test@example.com",
"password": "password123",
}
)
assert login_response.status_code == 200
access_token = login_response.json()["access_token"]
# Access protected route
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {access_token}"}
)
assert response.status_code == 200
```
## Best Practices
### DO
- Use async/await for I/O operations
- Use Pydantic for request/response validation
- Use FastAPI Depends for dependency injection
- Implement proper error handling with HTTPException
- Use status codes correctly (200, 201, 400, 401, 403, 404)
- Use background tasks for long-running operations
- Implement OAuth2 for third-party integrations
- Use JWT with proper expiration
- Write comprehensive tests with pytest
- Use type hints throughout
- Document APIs with OpenAPI
- Configure CORS properly
- Use environment variables for configuration
### DON'T
- Use synchronous database queries in routes
- Skip input validation (trust user input)
- Return database models directly (use schemas)
- Use plain HTTPException without status codes
- Mix business logic in routes (use services)
- Ignore error handling
- Use JWT without expiration (security risk)
- Hardcode configuration values
- Skip tests
- Use global state
- Run development server in production
## Anti-patterns
1. **Synchronous Operations** - Using blocking I/O in async routes
2. **No Validation** - Trusting all user input without validation
3. **SQL Injection** - Using raw SQL without parameterization
4. **Missing Depends** - Directly using database sessions
5. **Poor Error Handling** - Returning plain dicts instead of HTTPException
6. **No OAuth2** - Implementing custom auth instead of using FastAPI's support
7. **No Background Tasks** - Blocking on heavy operations
8. **Missing Type Hints** - Not adding type hints
9. **Hardcoded Secrets** - Committing API keys to code
10. **Insecure JWT** - No expiration or weak algorithms
11. **No Tests** - No unit or integration tests
12. **Global State** - Using global variables instead of dependency injection
13. **Poor CORS** - Allowing all origins in production
14. **No OpenAPI** - Not using auto-generated documentation
## Production Deployment
### Gunicorn Configuration
```python
# gunicorn_config.py
import multiprocessing
workers = multiprocessing.cpu_count() * 2 + 1
bind = "0.0.0.0:8000"
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
max_requests = 1000
max_requests_jitter = 50
timeout = 30
keepalive = 5
preload_app = True
reload = False
loglevel = "info"
accesslog = "-"
errorlog = "-"
```
### Dockerfile
```dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV PYTHONUNBUFFERED=1
EXPOSE 8000
CMD ["gunicorn", "gunicorn_config.py:app"]
```
### Docker Compose
```yaml
version: "3.8"
services:
web:
build: .
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://user:pass@db:5432/mydb
- SECRET_KEY=${SECRET_KEY}
- OAUTH_CLIENT_ID=${OAUTH_CLIENT_ID}
- OAUTH_CLIENT_SECRET=${OAUTH_CLIENT_SECRET}
depends_on:
- db
db:
image: postgres:15-alpine
environment:
POSTGRES_DB: mydb
POSTGRES_USER: appuser
POSTGRES_PASSWORD: secretpassword
volumes:
- postgres_data:/var/lib/postgresql/data
```
## Resources
- [FastAPI Documentation](https://fastapi.tiangolo.com/)
- [Pydantic Documentation](https://docs.pydantic.dev/)
- [FastAPI OAuth2](https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/)
- [FastAPI Background Tasks](https://fastapi.tiangolo.com/tutorial/background-tasks/)
- [FastAPI Testing](https://fastapi.tiangolo.com/tutorial/testing/)
- [SQLAlchemy Documentation](https://docs.sqlalchemy.org/)
- [Pytest Documentation](https://docs.pytest.org/)
Is this your skill, or is something wrong with this listing? Request removal or report an issue. Author removals are honored within 72 hours.
No comments yet. Be the first to comment!
