Files
gh-jeremylongshore-claude-c…/commands/fastapi-scaffold.md
2025-11-30 08:20:34 +08:00

674 lines
16 KiB
Markdown

---
description: Generate production-ready FastAPI REST API with async and authentication
shortcut: fas
category: backend
difficulty: intermediate
estimated_time: 5-10 minutes
---
# FastAPI Scaffold
Generates a complete FastAPI REST API boilerplate with async support, authentication, database integration, and testing setup.
## What This Command Does
**Generated Project:**
- FastAPI with Python 3.10+
- Async/await throughout
- JWT authentication
- Database integration (SQLAlchemy async)
- Pydantic models & validation
- Automatic OpenAPI docs
- Testing setup (Pytest + httpx)
- Docker configuration
- Example CRUD endpoints
**Output:** Complete API project ready for development
**Time:** 5-10 minutes
---
## Usage
```bash
# Generate full FastAPI API
/fastapi-scaffold "Task Management API"
# Shortcut
/fas "E-commerce API"
# With specific database
/fas "Blog API" --database postgresql
# With authentication type
/fas "Social API" --auth jwt --database postgresql
```
---
## Example Output
**Input:**
```
/fas "Task Management API" --database postgresql
```
**Generated Project Structure:**
```
task-api/
├── app/
│ ├── api/
│ │ ├── deps.py # Dependencies
│ │ └── v1/
│ │ ├── __init__.py
│ │ ├── auth.py # Auth endpoints
│ │ └── tasks.py # Task endpoints
│ ├── core/
│ │ ├── config.py # Settings
│ │ ├── security.py # JWT, password hashing
│ │ └── database.py # Database connection
│ ├── models/ # SQLAlchemy models
│ │ ├── user.py
│ │ └── task.py
│ ├── schemas/ # Pydantic schemas
│ │ ├── user.py
│ │ └── task.py
│ ├── services/ # Business logic
│ │ ├── auth.py
│ │ └── task.py
│ ├── db/
│ │ └── init_db.py # Database initialization
│ ├── main.py # FastAPI app
│ └── __init__.py
├── tests/
│ ├── conftest.py
│ ├── test_auth.py
│ └── test_tasks.py
├── alembic/ # Database migrations
│ ├── versions/
│ └── env.py
├── .env.example
├── .gitignore
├── requirements.txt
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
└── README.md
```
---
## Generated Files
### 1. **app/main.py** (Application Entry)
```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from app.api.v1 import auth, tasks
from app.core.config import settings
from app.core.database import engine
from app.models import Base
# Create database tables
Base.metadata.create_all(bind=engine)
app = FastAPI(
title=settings.PROJECT_NAME,
version="1.0.0",
openapi_url=f"{settings.API_V1_STR}/openapi.json",
docs_url=f"{settings.API_V1_STR}/docs",
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Security middleware
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=settings.ALLOWED_HOSTS
)
@app.get("/health")
async def health_check():
return {
"status": "ok",
"version": "1.0.0"
}
# Include routers
app.include_router(auth.router, prefix=f"{settings.API_V1_STR}/auth", tags=["auth"])
app.include_router(tasks.router, prefix=f"{settings.API_V1_STR}/tasks", tags=["tasks"])
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
```
### 2. **app/core/config.py** (Settings)
```python
from pydantic_settings import BaseSettings
from typing import List
class Settings(BaseSettings):
PROJECT_NAME: str = "Task API"
API_V1_STR: str = "/api/v1"
# Database
DATABASE_URL: str
# Security
SECRET_KEY: str
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 7 # 7 days
# CORS
ALLOWED_ORIGINS: List[str] = ["http://localhost:3000"]
ALLOWED_HOSTS: List[str] = ["*"]
class Config:
env_file = ".env"
case_sensitive = True
settings = Settings()
```
### 3. **app/core/security.py** (Authentication)
```python
from datetime import datetime, timedelta
from typing import Optional
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from app.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(
to_encode,
settings.SECRET_KEY,
algorithm=settings.ALGORITHM
)
return encoded_jwt
def decode_access_token(token: str) -> dict:
try:
payload = jwt.decode(
token,
settings.SECRET_KEY,
algorithms=[settings.ALGORITHM]
)
return payload
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
```
### 4. **app/core/database.py** (Database Setup)
```python
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
engine = create_engine(
settings.DATABASE_URL,
pool_pre_ping=True,
echo=False
)
SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=engine
)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
```
### 5. **app/models/user.py** (User Model)
```python
from sqlalchemy import Column, String, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid
from app.core.database import Base
class User(Base):
__tablename__ = "users"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
name = Column(String, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
tasks = relationship("Task", back_populates="owner", cascade="all, delete-orphan")
```
### 6. **app/models/task.py** (Task Model)
```python
from sqlalchemy import Column, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid
from app.core.database import Base
class Task(Base):
__tablename__ = "tasks"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
title = Column(String, nullable=False)
description = Column(String, nullable=True)
completed = Column(Boolean, default=False)
user_id = Column(String, ForeignKey("users.id", ondelete="CASCADE"))
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
owner = relationship("User", back_populates="tasks")
```
### 7. **app/schemas/user.py** (Pydantic Schemas)
```python
from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional
class UserBase(BaseModel):
email: EmailStr
name: str
class UserCreate(UserBase):
password: str
class UserUpdate(BaseModel):
name: Optional[str] = None
email: Optional[EmailStr] = None
class UserInDB(UserBase):
id: str
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class User(UserInDB):
pass
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
email: Optional[str] = None
```
### 8. **app/schemas/task.py** (Task Schemas)
```python
from pydantic import BaseModel
from datetime import datetime
from typing import Optional
class TaskBase(BaseModel):
title: str
description: Optional[str] = None
class TaskCreate(TaskBase):
pass
class TaskUpdate(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
completed: Optional[bool] = None
class TaskInDB(TaskBase):
id: str
completed: bool
user_id: str
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class Task(TaskInDB):
pass
```
### 9. **app/api/deps.py** (Dependencies)
```python
from typing import Generator
from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import oauth2_scheme, decode_access_token
from app.models.user import User
def get_current_user(
db: Session = Depends(get_db),
token: str = Depends(oauth2_scheme)
) -> User:
payload = decode_access_token(token)
email: str = payload.get("sub")
if email is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials"
)
user = db.query(User).filter(User.email == email).first()
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found"
)
return user
```
### 10. **app/api/v1/tasks.py** (Task Endpoints)
```python
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.core.database import get_db
from app.models.user import User
from app.models.task import Task as TaskModel
from app.schemas.task import Task, TaskCreate, TaskUpdate
router = APIRouter()
@router.get("/", response_model=List[Task])
async def list_tasks(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
skip: int = 0,
limit: int = 100
):
tasks = db.query(TaskModel)\
.filter(TaskModel.user_id == current_user.id)\
.offset(skip)\
.limit(limit)\
.all()
return tasks
@router.post("/", response_model=Task, status_code=status.HTTP_201_CREATED)
async def create_task(
task_in: TaskCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
task = TaskModel(
**task_in.dict(),
user_id=current_user.id
)
db.add(task)
db.commit()
db.refresh(task)
return task
@router.get("/{task_id}", response_model=Task)
async def get_task(
task_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
task = db.query(TaskModel)\
.filter(TaskModel.id == task_id, TaskModel.user_id == current_user.id)\
.first()
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found"
)
return task
@router.patch("/{task_id}", response_model=Task)
async def update_task(
task_id: str,
task_in: TaskUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
task = db.query(TaskModel)\
.filter(TaskModel.id == task_id, TaskModel.user_id == current_user.id)\
.first()
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found"
)
for field, value in task_in.dict(exclude_unset=True).items():
setattr(task, field, value)
db.commit()
db.refresh(task)
return task
@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(
task_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
task = db.query(TaskModel)\
.filter(TaskModel.id == task_id, TaskModel.user_id == current_user.id)\
.first()
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found"
)
db.delete(task)
db.commit()
```
### 11. **tests/test_tasks.py** (Pytest Tests)
```python
import pytest
from httpx import AsyncClient
from app.main import app
@pytest.mark.asyncio
async def test_create_task(client: AsyncClient, test_user_token):
response = await client.post(
"/api/v1/tasks/",
json={
"title": "Test Task",
"description": "Test description"
},
headers={"Authorization": f"Bearer {test_user_token}"}
)
assert response.status_code == 201
data = response.json()
assert data["title"] == "Test Task"
assert "id" in data
@pytest.mark.asyncio
async def test_list_tasks(client: AsyncClient, test_user_token):
response = await client.get(
"/api/v1/tasks/",
headers={"Authorization": f"Bearer {test_user_token}"}
)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
@pytest.mark.asyncio
async def test_create_task_unauthorized(client: AsyncClient):
response = await client.post(
"/api/v1/tasks/",
json={"title": "Test"}
)
assert response.status_code == 401
```
### 12. **requirements.txt**
```
fastapi==0.109.0
uvicorn[standard]==0.27.0
sqlalchemy==2.0.25
pydantic==2.5.3
pydantic-settings==2.1.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
alembic==1.13.1
psycopg2-binary==2.9.9
# Development
pytest==7.4.4
pytest-asyncio==0.23.3
httpx==0.26.0
black==23.12.1
isort==5.13.2
mypy==1.8.0
```
---
## Features
**Performance:**
- Async/await for high concurrency
- Background tasks support
- WebSocket support (optional)
- Automatic Pydantic validation
**Documentation:**
- Auto-generated OpenAPI (Swagger)
- ReDoc documentation
- Type hints throughout
**Database:**
- SQLAlchemy ORM with async support
- Alembic migrations
- Connection pooling
**Security:**
- JWT authentication
- Password hashing (bcrypt)
- CORS middleware
- Trusted host middleware
**Testing:**
- Pytest with async support
- Test fixtures
- Coverage reporting
---
## Getting Started
**1. Install dependencies:**
```bash
pip install -r requirements.txt
```
**2. Configure environment:**
```bash
cp .env.example .env
# Edit .env with your database URL and secrets
```
**3. Run database migrations:**
```bash
alembic upgrade head
```
**4. Start development server:**
```bash
uvicorn app.main:app --reload
```
**5. View API docs:**
- Swagger UI: http://localhost:8000/api/v1/docs
- ReDoc: http://localhost:8000/api/v1/redoc
**6. Run tests:**
```bash
pytest
```
---
## Related Commands
- `/express-api-scaffold` - Generate Express.js boilerplate
- Backend Architect (agent) - Architecture review
- API Builder (agent) - API design guidance
---
**Build high-performance APIs. Scale effortlessly. Deploy confidently.**