Intermediate

Authentication & Security

Secure your ML API with API key authentication, JWT tokens, OAuth2, and rate limiting to prevent abuse and control access.

API Key Authentication

Python

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

app = FastAPI()
api_key_header = APIKeyHeader(name="X-API-Key")

VALID_API_KEYS = {os.environ["API_KEY"]}

async def verify_api_key(api_key: str = Security(api_key_header)):
    if api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

@app.post("/predict")
async def predict(input: TextInput, api_key: str = Depends(verify_api_key)):
    return {"prediction": model.predict(input.text)}

JWT Token Authentication

Python

from jose import JWTError, jwt
from fastapi.security import OAuth2PasswordBearer
from datetime import datetime, timedelta

SECRET_KEY = os.environ["JWT_SECRET"]
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(hours=1))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")

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

@app.post("/predict")
async def predict(input: TextInput, user: str = Depends(get_current_user)):
    return {"prediction": model.predict(input.text), "user": user}

Rate Limiting

Python - Using slowapi

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/predict")
@limiter.limit("10/minute")
async def predict(request: Request, input: TextInput):
    return {"prediction": model.predict(input.text)}

CORS Configuration

Python

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["POST"],
    allow_headers=["*"],
)

⚠

Security checklist: Never hardcode secrets. Use environment variables. Always validate and sanitize inputs. Enable HTTPS in production. Log authentication failures for monitoring.

What's Next?

In our final lesson, we cover Docker deployment, health checks, monitoring, testing, and production best practices.

← PreviousStreaming Next →Best Practices