fastapi-expert

Expert guidance for FastAPI - modern, fast Python web framework for building APIs with automatic OpenAPI documentation and type safety.

Safety Notice

This listing is imported from skills.sh public index metadata. Review upstream SKILL.md and repository scripts before running.

Copy this and send it to your AI assistant to learn

Install skill "fastapi-expert" with this command: npx skills add personamanagmentlayer/pcl/personamanagmentlayer-pcl-fastapi-expert

FastAPI Expert

Expert guidance for FastAPI - modern, fast Python web framework for building APIs with automatic OpenAPI documentation and type safety.

Core Concepts

FastAPI Features

  • Fast performance (Starlette + Pydantic)

  • Automatic OpenAPI/Swagger docs

  • Type hints and validation

  • Async/await support

  • Dependency injection

  • OAuth2 and JWT

  • WebSocket support

Key Components

  • Path operations (routes)

  • Request/response models (Pydantic)

  • Dependency injection

  • Middleware

  • Background tasks

  • Testing with TestClient

Basic FastAPI Application

from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel, EmailStr, Field from typing import List, Optional import uvicorn

app = FastAPI( title="My API", description="Production-ready FastAPI", version="1.0.0" )

Models

class UserCreate(BaseModel): email: EmailStr name: str = Field(..., min_length=2, max_length=100) age: Optional[int] = Field(None, ge=0, le=150)

class UserResponse(BaseModel): id: int email: str name: str

class Config:
    from_attributes = True

Routes

@app.post("/users", response_model=UserResponse, status_code=201) async def create_user(user: UserCreate): # Create user in database db_user = await db.users.create(**user.dict()) return db_user

@app.get("/users/{user_id}", response_model=UserResponse) async def get_user(user_id: int): user = await db.users.get(user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user

@app.get("/users", response_model=List[UserResponse]) async def list_users(skip: int = 0, limit: int = 100): return await db.users.find_many(skip=skip, limit=limit)

if name == "main": uvicorn.run(app, host="0.0.0.0", port=8000)

Dependency Injection

from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from sqlalchemy.ext.asyncio import AsyncSession

Database dependency

async def get_db() -> AsyncSession: async with async_session() as session: yield session

Auth dependency

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user( token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db) ): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials" )

payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: int = payload.get("sub")
if user_id is None:
    raise credentials_exception

user = await db.get(User, user_id)
if user is None:
    raise credentials_exception

return user

Use dependencies

@app.get("/me", response_model=UserResponse) async def read_users_me(current_user: User = Depends(get_current_user)): return current_user

@app.post("/posts") async def create_post( post: PostCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db) ): db_post = Post(**post.dict(), user_id=current_user.id) db.add(db_post) await db.commit() return db_post

Authentication

from datetime import datetime, timedelta from jose import JWTError, jwt from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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: timedelta = None): to_encode = data.copy() expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token") async def login( form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db) ): user = await authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException(status_code=401, detail="Invalid credentials")

access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}

Database Integration (SQLAlchemy)

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker, declarative_base from sqlalchemy import Column, Integer, String, Boolean

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True) async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Base = declarative_base()

class User(Base): tablename = "users"

id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean, default=True)

CRUD operations

async def get_user(db: AsyncSession, user_id: int): return await db.get(User, user_id)

async def create_user(db: AsyncSession, user: UserCreate): hashed_password = get_password_hash(user.password) db_user = User(email=user.email, hashed_password=hashed_password) db.add(db_user) await db.commit() await db.refresh(db_user) return db_user

Request Validation

from pydantic import BaseModel, validator, root_validator from typing import Optional from datetime import date

class PostCreate(BaseModel): title: str = Field(..., min_length=5, max_length=200) content: str = Field(..., min_length=10) published: bool = False tags: List[str] = Field(default_factory=list, max_items=10)

@validator('tags')
def validate_tags(cls, v):
    return [tag.lower().strip() for tag in v]

@validator('title')
def validate_title(cls, v):
    if any(word in v.lower() for word in ['spam', 'xxx']):
        raise ValueError('Invalid content')
    return v

Background Tasks

from fastapi import BackgroundTasks

def send_email(email: str, message: str): # Send email print(f"Sending email to {email}: {message}")

@app.post("/send-notification") async def send_notification( email: str, background_tasks: BackgroundTasks ): background_tasks.add_task(send_email, email, "Welcome!") return {"message": "Notification sent"}

Testing

from fastapi.testclient import TestClient import pytest

client = TestClient(app)

def test_create_user(): response = client.post( "/users", json={"email": "test@example.com", "name": "Test User"} ) assert response.status_code == 201 assert response.json()["email"] == "test@example.com"

def test_get_user(): response = client.get("/users/1") assert response.status_code == 200 assert "email" in response.json()

@pytest.mark.asyncio async def test_async_endpoint(): async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/users") assert response.status_code == 200

Best Practices

  • Use Pydantic models for validation

  • Implement proper error handling

  • Use dependency injection

  • Add rate limiting

  • Enable CORS properly

  • Use async/await for I/O

  • Document with docstrings

  • Add proper logging

Resources

Source Transparency

This detail page is rendered from real SKILL.md content. Trust labels are metadata-based hints, not a safety guarantee.

Open in GitHubOpen in ClawHub

Related Skills

Related by shared tags or category signals.

Coding

python-expert

No summary provided by upstream source.

Repository SourceNeeds Review
76-personamanagmentlayer
Coding

devops-expert

No summary provided by upstream source.

Repository SourceNeeds Review
47-personamanagmentlayer
Coding

code-review-expert

No summary provided by upstream source.

Repository SourceNeeds Review
46-personamanagmentlayer
Coding

typescript-expert

No summary provided by upstream source.

Repository SourceNeeds Review
40-personamanagmentlayer