🌐 Professional API Development

📌 What You'll Learn

• •REST API principles and design
• •Building APIs with FastAPI and Flask
• •Authentication (JWT, API Keys, OAuth)
• •Request validation and error handling
• •API documentation with OpenAPI/Swagger
• •Production best practices

---

🔍 What is a REST API?

REST (Representational State Transfer) is an architectural style for building web APIs. APIs allow different software systems to communicate.

REST Principles

1. •Resources - Everything is a resource (users, posts, products)
2. •URLs identify resources - /api/users/123
3. •HTTP methods define actions - GET, POST, PUT, DELETE
4. •Stateless - Each request contains all needed information
5. •JSON responses - Standard data format

HTTP Methods

HTTP Status Codes

---

⚡ FastAPI - Modern Python API Framework

FastAPI is a modern, fast framework with automatic documentation.

pip install fastapi uvicorn

Basic FastAPI Application

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

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

# Data model using Pydantic
class User(BaseModel):
    id: Optional[int] = None
    name: str
    email: str
    age: Optional[int] = None

class UserCreate(BaseModel):
    name: str
    email: str
    age: Optional[int] = None

# In-memory database (use real DB in production)
users_db: dict[int, User] = {}
next_id = 1

# GET - List all users
@app.get("/users", response_model=list[User])
def get_users(skip: int = 0, limit: int = 10):
    """Get all users with pagination."""
    return list(users_db.values())[skip:skip + limit]

# GET - Get single user
@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int):
    """Get a user by ID."""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return users_db[user_id]

# POST - Create user
@app.post("/users", response_model=User, status_code=201)
def create_user(user: UserCreate):
    """Create a new user."""
    global next_id
    new_user = User(id=next_id, **user.dict())
    users_db[next_id] = new_user
    next_id += 1
    return new_user

# PUT - Update user
@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, user: UserCreate):
    """Update a user."""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    updated_user = User(id=user_id, **user.dict())
    users_db[user_id] = updated_user
    return updated_user

# DELETE - Delete user
@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int):
    """Delete a user."""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    del users_db[user_id]

# Run: uvicorn main:app --reload
# Docs: http://127.0.0.1:8000/docs

Path and Query Parameters

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

app = FastAPI()

@app.get("/items/{item_id}")
def get_item(
    # Path parameter with validation
    item_id: int = Path(..., title="Item ID", ge=1),

    # Query parameters
    q: Optional[str] = Query(None, min_length=3, max_length=50),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, le=100)
):
    """
    Get an item with optional search query.

    - **item_id**: The item's unique ID (required)
    - **q**: Optional search query (3-50 chars)
    - **skip**: Number of items to skip (default: 0)
    - **limit**: Max items to return (default: 10, max: 100)
    """
    return {
        "item_id": item_id,
        "query": q,
        "skip": skip,
        "limit": limit
    }

# GET /items/42?q=search&skip=0&limit=20

Request Body Validation

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, validator, EmailStr
from typing import Optional
from datetime import datetime

app = FastAPI()

class ProductCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    price: float = Field(..., gt=0, description="Price must be positive")
    quantity: int = Field(default=0, ge=0)
    category: str

    @validator("category")
    def validate_category(cls, v):
        allowed = ["electronics", "clothing", "food", "books"]
        if v.lower() not in allowed:
            raise ValueError(f"Category must be one of: {allowed}")
        return v.lower()

    class Config:
        schema_extra = {
            "example": {
                "name": "Laptop",
                "description": "A powerful laptop",
                "price": 999.99,
                "quantity": 10,
                "category": "electronics"
            }
        }

class Product(ProductCreate):
    id: int
    created_at: datetime

@app.post("/products", response_model=Product, status_code=201)
def create_product(product: ProductCreate):
    # Pydantic automatically validates the request body
    # Invalid data returns 422 with detailed error messages
    return Product(
        id=1,
        created_at=datetime.now(),
        **product.dict()
    )

---

🔐 Authentication

API Key Authentication

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader

app = FastAPI()

API_KEY = "your-secret-api-key"
api_key_header = APIKeyHeader(name="X-API-Key")

def verify_api_key(api_key: str = Security(api_key_header)):
    if api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return api_key

@app.get("/protected")
def protected_route(api_key: str = Depends(verify_api_key)):
    return {"message": "You have access!"}

# curl -H "X-API-Key: your-secret-api-key" http://localhost:8000/protected

JWT Authentication

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel

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

SECRET_KEY = "your-secret-key"  # Use environment variable in production!
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

# Fake user database
fake_users_db = {
    "alice": {
        "username": "alice",
        "hashed_password": pwd_context.hash("secret123"),
        "email": "alice@example.com"
    }
}

class Token(BaseModel):
    access_token: str
    token_type: str

class User(BaseModel):
    username: str
    email: str

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    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)

def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user_data = fake_users_db.get(username)
    if user_data is None:
        raise credentials_exception
    return User(**user_data)

@app.post("/token", response_model=Token)
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    access_token = create_access_token(
        data={"sub": user["username"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

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

---

🧪 Error Handling

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

# Custom exception
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# Exception handler
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "error": "item_not_found",
            "message": f"Item {exc.item_id} not found",
            "path": str(request.url)
        }
    )

# Generic exception handler
@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "error": "internal_server_error",
            "message": "An unexpected error occurred"
        }
    )

@app.get("/items/{item_id}")
def get_item(item_id: int):
    items = {1: "Item One", 2: "Item Two"}
    if item_id not in items:
        raise ItemNotFoundError(item_id)
    return {"id": item_id, "name": items[item_id]}

---

🌶️ Flask - Simple and Flexible

Flask is a lightweight framework, great for simple APIs.

pip install flask

Basic Flask API

from flask import Flask, jsonify, request, abort

app = Flask(__name__)

# In-memory database
users = {
    1: {"id": 1, "name": "Alice", "email": "alice@example.com"},
    2: {"id": 2, "name": "Bob", "email": "bob@example.com"}
}
next_id = 3

@app.route("/users", methods=["GET"])
def get_users():
    return jsonify(list(users.values()))

@app.route("/users/<int:user_id>", methods=["GET"])
def get_user(user_id):
    user = users.get(user_id)
    if not user:
        abort(404)
    return jsonify(user)

@app.route("/users", methods=["POST"])
def create_user():
    global next_id
    data = request.get_json()

    if not data or "name" not in data or "email" not in data:
        abort(400)

    user = {
        "id": next_id,
        "name": data["name"],
        "email": data["email"]
    }
    users[next_id] = user
    next_id += 1

    return jsonify(user), 201

@app.route("/users/<int:user_id>", methods=["DELETE"])
def delete_user(user_id):
    if user_id not in users:
        abort(404)
    del users[user_id]
    return "", 204

# Error handlers
@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "Not found"}), 404

@app.errorhandler(400)
def bad_request(error):
    return jsonify({"error": "Bad request"}), 400

if __name__ == "__main__":
    app.run(debug=True)

---

📚 API Documentation

FastAPI automatically generates OpenAPI documentation:

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

app = FastAPI(
    title="My API",
    description="API for managing users and items",
    version="1.0.0",
    contact={
        "name": "API Support",
        "email": "support@example.com"
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT"
    }
)

class Item(BaseModel):
    """A product item."""
    name: str = Field(..., description="The item name", example="Laptop")
    price: float = Field(..., gt=0, description="Price in USD", example=999.99)

    class Config:
        schema_extra = {
            "example": {
                "name": "Laptop",
                "price": 999.99
            }
        }

@app.get(
    "/items",
    summary="List all items",
    description="Retrieve a list of all items with optional pagination.",
    response_description="List of items",
    tags=["Items"]
)
def list_items(
    skip: int = Query(0, description="Items to skip"),
    limit: int = Query(10, description="Max items to return")
):
    """
    Get all items from the database.

    - **skip**: Number of items to skip (for pagination)
    - **limit**: Maximum number of items to return
    """
    return []

# Access docs at:
# - http://localhost:8000/docs (Swagger UI)
# - http://localhost:8000/redoc (ReDoc)

---

🔧 Production Best Practices

1. Use Environment Variables

import os
from pydantic import BaseSettings

class Settings(BaseSettings):
    database_url: str
    secret_key: str
    debug: bool = False

    class Config:
        env_file = ".env"

settings = Settings()

2. Add CORS

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],  # Or ["*"] for all
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

3. Rate Limiting

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/limited")
@limiter.limit("5/minute")
def limited_endpoint(request: Request):
    return {"message": "This endpoint is rate limited"}

4. Structured Logging

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_obj = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module
        }
        return json.dumps(log_obj)

logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

---

📋 API Design Checklist

• • Use nouns for resources (/users, not /getUsers)
• • Use proper HTTP methods (GET, POST, PUT, DELETE)
• • Return appropriate status codes
• • Version your API (/api/v1/...)
• • Paginate list endpoints
• • Validate all input
• • Return consistent error format
• • Document all endpoints
• • Implement authentication
• • Add rate limiting
• • Enable CORS appropriately
• • Log all requests
• • Use HTTPS in production

---

🎯 Next Steps

After learning API development, proceed to 28_docker_deployment to learn how to containerize and deploy your applications!