メニュー
FastAPI, REST APIs, GraphQL, data service design, and API best practices
Apache Spark, Hadoop, distributed computing, and large-scale data processing for petabyte-scale workloads
Portfolio building, technical interviews, job search strategies, and continuous learning
GitHub Actions, GitLab CI, Jenkins, and automated deployment pipelines
AWS, GCP, Azure data platforms, infrastructure as code, and cloud-native data solutions
Docker, Kubernetes, container orchestration, and cloud-native deployment for data applications
Data pipeline architecture, ETL/ELT patterns, data modeling, and production data platform design
| name | api-development |
| description | FastAPI, REST APIs, GraphQL, data service design, and API best practices |
| sasmp_version | 1.3.0 |
| bonded_agent | 02-backend-developer |
| bond_type | PRIMARY_BOND |
| skill_version | 2.0.0 |
| last_updated | 2025-01 |
| complexity | intermediate |
| estimated_mastery_hours | 100 |
| prerequisites | ["python-programming","sql-databases"] |
| unlocks | ["containerization","monitoring-observability"] |
Production-grade API development with FastAPI, REST best practices, and data service patterns.
from fastapi import FastAPI, HTTPException, Depends, Query
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
import uvicorn
app = FastAPI(title="Data API", version="1.0.0")
# Pydantic models for validation
class DataRecord(BaseModel):
id: str = Field(..., description="Unique identifier")
value: float = Field(..., ge=0, description="Non-negative value")
timestamp: datetime = Field(default_factory=datetime.utcnow)
metadata: Optional[dict] = None
class Config:
json_schema_extra = {
"example": {"id": "rec-001", "value": 42.5, "metadata": {"source": "sensor-1"}}
}
class DataResponse(BaseModel):
data: list[DataRecord]
total: int
page: int
page_size: int
@app.get("/data", response_model=DataResponse)
async def get_data(
page: int = Query(1, ge=1),
page_size: int = Query(100, ge=1, le=1000),
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None
):
"""Retrieve paginated data records with optional date filtering."""
# Query data with pagination
records = query_database(page, page_size, start_date, end_date)
total = count_records(start_date, end_date)
return DataResponse(data=records, total=total, page=page, page_size=page_size)
@app.post("/data", status_code=201)
async def create_data(record: DataRecord):
"""Create a new data record."""
try:
save_to_database(record)
return {"status": "created", "id": record.id}
except DuplicateKeyError:
raise HTTPException(status_code=409, detail="Record already exists")
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
security = HTTPBearer()
# Database session dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Authentication dependency
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)) -> dict:
token = credentials.credentials
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
return payload
except jwt.InvalidTokenError:
raise HTTPException(status_code=401, detail="Invalid token")
# Rate limiting dependency
class RateLimiter:
def __init__(self, requests_per_minute: int = 60):
self.requests_per_minute = requests_per_minute
async def __call__(self, request: Request):
client_ip = request.client.host
if is_rate_limited(client_ip, self.requests_per_minute):
raise HTTPException(status_code=429, detail="Rate limit exceeded")
# Usage
@app.get("/protected")
async def protected_endpoint(
user: dict = Depends(verify_token),
db: Session = Depends(get_db),
_: None = Depends(RateLimiter(100))
):
return {"user": user["sub"]}
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from pydantic import ValidationError
app = FastAPI()
class APIError(Exception):
def __init__(self, code: str, message: str, status_code: int = 400):
self.code = code
self.message = message
self.status_code = status_code
@app.exception_handler(APIError)
async def api_error_handler(request: Request, exc: APIError):
return JSONResponse(
status_code=exc.status_code,
content={"error": {"code": exc.code, "message": exc.message}}
)
@app.exception_handler(ValidationError)
async def validation_error_handler(request: Request, exc: ValidationError):
return JSONResponse(
status_code=422,
content={"error": {"code": "VALIDATION_ERROR", "details": exc.errors()}}
)
@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception):
logger.error(f"Unhandled error: {exc}", exc_info=True)
return JSONResponse(
status_code=500,
content={"error": {"code": "INTERNAL_ERROR", "message": "An unexpected error occurred"}}
)
from fastapi import BackgroundTasks
from celery import Celery
# Simple background tasks
@app.post("/reports")
async def generate_report(background_tasks: BackgroundTasks, report_id: str):
background_tasks.add_task(process_report, report_id)
return {"status": "processing", "report_id": report_id}
def process_report(report_id: str):
# Long-running task
time.sleep(60)
save_report(report_id)
# Celery for distributed tasks
celery_app = Celery('tasks', broker='redis://localhost:6379')
@celery_app.task
def async_etl_job(job_id: str):
run_etl_pipeline(job_id)
@app.post("/jobs")
async def start_job(job_id: str):
task = async_etl_job.delay(job_id)
return {"task_id": task.id, "status": "queued"}
| Tool | Purpose | Version (2025) |
|---|---|---|
| FastAPI | Modern API framework | 0.109+ |
| Pydantic | Data validation | 2.5+ |
| SQLAlchemy | Database ORM | 2.0+ |
| Celery | Task queue | 5.3+ |
| httpx | Async HTTP client | 0.27+ |
| pytest | Testing | 8.0+ |
| Issue | Symptoms | Root Cause | Fix |
|---|---|---|---|
| 422 Error | Validation failed | Invalid request data | Check request schema |
| Slow Response | High latency | Blocking I/O | Use async, background tasks |
| Connection Pool | DB timeouts | Pool exhausted | Increase pool, use limits |
| Memory Leak | Growing memory | Unclosed connections | Use context managers |
# ✅ DO: Use Pydantic for validation
class CreateUser(BaseModel):
email: EmailStr
name: str = Field(..., min_length=1, max_length=100)
# ✅ DO: Version your API
app = FastAPI()
v1 = APIRouter(prefix="/v1")
# ✅ DO: Use proper HTTP status codes
# 201 Created, 204 No Content, 404 Not Found
# ✅ DO: Document with OpenAPI
@app.get("/users/{user_id}", summary="Get user by ID", tags=["users"])
# ❌ DON'T: Return 200 for errors
# ❌ DON'T: Expose internal errors to clients
# ❌ DON'T: Skip input validation
Skill Certification Checklist: