⚡ Python 11 — FastAPI Framework

1. Setup & Hello World

pip install fastapi uvicorn[standard]
# Optional: pip install sqlalchemy pydantic[email] python-jose python-multipart

# main.py
from fastapi import FastAPI

app = FastAPI(title="My API", version="1.0.0")

@app.get("/")
def root():
    return {"message": "Hello FastAPI!"}

@app.get("/health")
def health():
    return {"status": "ok"}

# Run (hot reload)
uvicorn main:app --reload
# http://127.0.0.1:8000        → API
# http://127.0.0.1:8000/docs   → Swagger UI (auto!)
# http://127.0.0.1:8000/redoc  → ReDoc

2. Pydantic Models (Input/Output Validation)

from pydantic import BaseModel, EmailStr, validator
from typing import Optional
from datetime import datetime

# Request body schema
class PostCreate(BaseModel):
    title:   str
    content: str
    published: bool = False

# Response schema
class PostResponse(BaseModel):
    id:       int
    title:    str
    content:  str
    published: bool
    created:  datetime

    class Config:
        from_attributes = True   # ORM model ကို dict convert လုပ်နိုင်

# User schema with validation
class UserCreate(BaseModel):
    username: str
    email:    EmailStr
    password: str

    @validator('username')
    def username_must_not_empty(cls, v):
        if len(v) < 3:
            raise ValueError('Username must be at least 3 characters')
        return v.strip()

# Nested schema
class AuthorBase(BaseModel):
    id: int
    name: str

class PostWithAuthor(BaseModel):
    id:     int
    title:  str
    author: AuthorBase

3. Path, Query & Body Parameters

from fastapi import FastAPI, Path, Query, Body
from typing import Optional, List

app = FastAPI()

# Path parameter (URL variable)
@app.get("/users/{user_id}")
def get_user(user_id: int = Path(..., gt=0, description="User ID")):
    return {"user_id": user_id}

# Query parameters: /posts?published=true&page=1&limit=10
@app.get("/posts")
def get_posts(
    published: bool = True,
    page:      int  = Query(1,  ge=1),
    limit:     int  = Query(10, ge=1, le=100),
    search:    Optional[str] = None,
):
    return {"published": published, "page": page,
            "limit": limit, "search": search}

# Request body
class PostCreate(BaseModel):
    title:   str
    content: str

@app.post("/posts", status_code=201)
def create_post(post: PostCreate):
    # post.title, post.content available
    return {"id": 1, "title": post.title, "content": post.content}

# Path + body + query combined
@app.put("/posts/{post_id}")
def update_post(
    post_id: int,
    post:    PostCreate,
    notify:  bool = False,
):
    return {"id": post_id, "updated": post, "notified": notify}

4. CRUD API — Full Example (In-Memory)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

app = FastAPI()

# Schemas
class PostCreate(BaseModel):
    title:   str
    content: str

class Post(PostCreate):
    id:      int
    created: datetime

# In-memory DB
db: List[Post] = []
counter = 0

@app.get("/api/posts", response_model=List[Post])
def list_posts():
    return db

@app.get("/api/posts/{post_id}", response_model=Post)
def get_post(post_id: int):
    post = next((p for p in db if p.id == post_id), None)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    return post

@app.post("/api/posts", response_model=Post, status_code=201)
def create_post(data: PostCreate):
    global counter
    counter += 1
    post = Post(id=counter, created=datetime.utcnow(), **data.dict())
    db.append(post)
    return post

@app.put("/api/posts/{post_id}", response_model=Post)
def update_post(post_id: int, data: PostCreate):
    for i, p in enumerate(db):
        if p.id == post_id:
            db[i] = Post(id=post_id, created=p.created, **data.dict())
            return db[i]
    raise HTTPException(status_code=404, detail="Post not found")

@app.delete("/api/posts/{post_id}")
def delete_post(post_id: int):
    global db
    before = len(db)
    db = [p for p in db if p.id != post_id]
    if len(db) == before:
        raise HTTPException(status_code=404, detail="Post not found")
    return {"message": "Deleted"}

5. Database (SQLAlchemy + SQLite/PostgreSQL)

pip install sqlalchemy

# database.py
from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

DATABASE_URL = "sqlite:///./blog.db"
# PostgreSQL: "postgresql://user:pass@localhost/dbname"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()

# ORM Model
class PostModel(Base):
    __tablename__ = "posts"
    id        = Column(Integer, primary_key=True, index=True)
    title     = Column(String(200), nullable=False)
    content   = Column(String)
    published = Column(Boolean, default=False)
    created   = Column(DateTime, default=datetime.utcnow)

Base.metadata.create_all(bind=engine)

# main.py - Dependency Injection
from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/api/posts")
def list_posts(db: Session = Depends(get_db)):
    return db.query(PostModel).filter(PostModel.published == True).all()

@app.post("/api/posts", status_code=201)
def create_post(data: PostCreate, db: Session = Depends(get_db)):
    post = PostModel(**data.dict())
    db.add(post)
    db.commit()
    db.refresh(post)
    return post

6. JWT Authentication

pip install python-jose[cryptography] passlib[bcrypt]

# auth.py
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

SECRET_KEY = "your-secret-key"
ALGORITHM  = "HS256"
TOKEN_EXPIRE_MINUTES = 60

pwd_context   = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_token(data: dict) -> str:
    payload = data.copy()
    payload["exp"] = datetime.utcnow() + timedelta(minutes=TOKEN_EXPIRE_MINUTES)
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user_id

# Routes
@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
    # Verify user from DB here
    if form.username == "test" and form.password == "secret":
        token = create_token({"sub": form.username})
        return {"access_token": token, "token_type": "bearer"}
    raise HTTPException(status_code=401, detail="Wrong credentials")

@app.get("/me")
def me(user_id: str = Depends(get_current_user)):
    return {"user": user_id}

7. Middleware, CORS & Background Tasks

from fastapi.middleware.cors import CORSMiddleware
from fastapi import BackgroundTasks

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "http://localhost:8080"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Background task (after response)
def send_email(email: str, subject: str):
    import time; time.sleep(2)
    print(f"Email sent to {email}: {subject}")

@app.post("/register")
def register(bg: BackgroundTasks):
    bg.add_task(send_email, "user@example.com", "Welcome!")
    return {"message": "Registered (email sending in background)"}

8. FastAPI vs Flask vs Django — Comparison

9. Project Structure (Recommended)

myapi/
├── main.py             ← FastAPI app entry
├── database.py         ← DB engine + session
├── models.py           ← SQLAlchemy ORM models
├── schemas.py          ← Pydantic request/response schemas
├── auth.py             ← JWT, password utils
├── routers/
│   ├── posts.py        ← POST routes (APIRouter)
│   └── users.py        ← USER routes (APIRouter)
├── .env                ← secrets (SECRET_KEY, DB URL)
└── requirements.txt

# main.py — router registration
from fastapi import FastAPI
from routers import posts, users

app = FastAPI()
app.include_router(posts.router, prefix="/api/posts", tags=["Posts"])
app.include_router(users.router, prefix="/api/users", tags=["Users"])