APIs Python — Cours Complet

1. REST APIs avec FastAPI

REST — Principes Fondamentaux

REST (Representational State Transfer) est basé sur :

• Ressources identifiées par URLs
• Verbes HTTP (GET, POST, PUT, PATCH, DELETE)
• Stateless : chaque requête contient toutes les infos
• Représentations : JSON, XML, etc.

API REST Complète

from fastapi import FastAPI, HTTPException, Query, Path, status
from pydantic import BaseModel, Field
from typing import Optional, Annotated

app = FastAPI(title="REST API Example", version="2.0.0")

# Pagination standard
class PaginationParams:
    def __init__(
        self,
        skip: int = Query(0, ge=0, description="Records to skip"),
        limit: int = Query(20, ge=1, le=100, description="Page size"),
    ):
        self.skip = skip
        self.limit = limit

Pagination = Annotated[PaginationParams, Depends()]

# API Versioning via headers
from fastapi import Header

@app.get("/api/v2/users")
async def list_users_v2(
    pagination: Pagination,
    x_api_version: str = Header("2"),
) -> list[dict]:
    return [{"id": 1, "name": "Alice"}]  # v2 response format

Versioning d'API

# Stratégies de versioning

# 1. URL Path (recommandé)
app = FastAPI()
v1 = APIRouter(prefix="/api/v1")
v2 = APIRouter(prefix="/api/v2")

@v1.get("/users")
async def list_users_v1(): ...

@v2.get("/users")
async def list_users_v2(): ...

# 2. Header
@app.get("/users")
async def list_users(
    accept_version: str = Header("v1"),
): ...

# 3. Query parameter
@app.get("/users")
async def list_users(version: str = "v1"): ...

Error Handling — Problem JSON (RFC 7807)

from fastapi import Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

class ProblemDetail(BaseModel):
    type: str
    title: str
    status: int
    detail: str
    instance: str

@app.exception_handler(HTTPException)
async def problem_exception_handler(request: Request, exc: HTTPException):
    problem = ProblemDetail(
        type="https://example.com/errors/not-found",
        title="Resource not found",
        status=exc.status_code,
        detail=exc.detail,
        instance=str(request.url),
    )
    return JSONResponse(
        status_code=exc.status_code,
        content=problem.model_dump(),
        headers={"Content-Type": "application/problem+json"},
    )

2. GraphQL

Strawberry — GraphQL Moderne

import strawberry
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
from typing import Optional

# Types
@strawberry.type
class User:
    id: strawberry.ID
    name: str
    email: str

@strawberry.type
class Post:
    id: strawberry.ID
    title: str
    content: str
    author: User

# Query
@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: strawberry.ID) -> Optional[User]:
        # DB lookup
        return User(id=id, name="Alice", email="alice@example.com")

    @strawberry.field
    def users(self) -> list[User]:
        return [
            User(id="1", name="Alice", email="alice@example.com"),
            User(id="2", name="Bob", email="bob@example.com"),
        ]

# Mutation
@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, name: str, email: str) -> User:
        # Create in DB
        return User(id="3", name=name, email=email)

# Subscription
@strawberry.type
class Subscription:
    @strawberry.subscription
    async def user_created(self) -> strawberry.AsyncIterator[User]:
        while True:
            await asyncio.sleep(1)
            yield User(id="new", name="New User", email="new@example.com")

schema = strawberry.Schema(query=Query, mutation=Mutation, subscription=Subscription)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

Graphene — GraphQL Traditionnel

import graphene

class User(graphene.ObjectType):
    id = graphene.ID()
    name = graphene.String()
    email = graphene.String()

class Query(graphene.ObjectType):
    user = graphene.Field(User, id=graphene.ID())
    users = graphene.List(User)

    def resolve_user(self, info, id):
        return User(id=id, name="Alice", email="alice@example.com")

    def resolve_users(self, info):
        return [User(id="1", name="Alice", email="alice@example.com")]

class CreateUser(graphene.Mutation):
    class Arguments:
        name = graphene.String()
        email = graphene.String()

    user = graphene.Field(User)

    def mutate(self, info, name, email):
        return CreateUser(user=User(id="3", name=name, email=email))

class Mutation(graphene.ObjectType):
    create_user = CreateUser.Field()

schema = graphene.Schema(query=Query, mutation=Mutation)

3. WebSocket

FastAPI WebSocket

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Set
import json

app = FastAPI()

class ChatManager:
    def __init__(self):
        self.connections: Set[WebSocket] = set()

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.connections.add(ws)

    def disconnect(self, ws: WebSocket):
        self.connections.discard(ws)

    async def broadcast(self, message: dict):
        dead = set()
        for conn in self.connections:
            try:
                await conn.send_json(message)
            except Exception:
                dead.add(conn)
        self.connections -= dead

manager = ChatManager()

@app.websocket("/ws/chat")
async def chat(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            data = await ws.receive_json()
            await manager.broadcast({
                "type": "message",
                "user": data.get("user", "anonymous"),
                "text": data["text"],
            })
    except WebSocketDisconnect:
        manager.disconnect(ws)
        await manager.broadcast({"type": "user_left"})

Django Channels

# consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = f"chat_{self.room_name}"

        await self.channel_layer.group_add(self.room_group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(self.room_group_name, self.channel_name)

    async def receive(self, text_data):
        data = json.loads(text_data)
        await self.channel_layer.group_send(
            self.room_group_name,
            {"type": "chat.message", "message": data["message"]},
        )

    async def chat_message(self, event):
        await self.send(text_data=json.dumps({"message": event["message"]}))

4. Rate Limiting

"""Rate limiting avec slowapi."""

from fastapi import FastAPI, Request
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 = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/unlimited")
async def unlimited():
    return {"message": "No limit"}

@app.get("/limited")
@limiter.limit("5/minute")
async def limited(request: Request):
    return {"message": "Limited to 5 per minute"}

# Rate limiting par utilisateur
@limiter.limit("100/hour")
@app.get("/api/users")
async def list_users(request: Request):
    ...

5. Authentification

JWT

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

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str = Depends(oauth2_scheme)) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

OAuth2

from authlib.integrations.starlette_client import OAuth
from starlette.config import Config

config = Config(".env")
oauth = OAuth(config)

oauth.register(
    name="google",
    client_id=config("GOOGLE_CLIENT_ID"),
    client_secret=config("GOOGLE_CLIENT_SECRET"),
    authorize_url="https://accounts.google.com/o/oauth2/auth",
    authorize_params=None,
    access_token_url="https://accounts.google.com/o/oauth2/token",
    client_kwargs={"scope": "openid email profile"},
)

@app.get("/auth/google")
async def google_login(request: Request):
    redirect = request.url_for("auth_callback")
    return await oauth.google.authorize_redirect(request, redirect)

@app.get("/auth/callback")
async def auth_callback(request: Request):
    token = await oauth.google.authorize_access_token(request)
    user_info = await oauth.google.parse_id_token(request, token)
    return {"user": user_info}

API Keys

from fastapi import Header, HTTPException

API_KEYS = {"key-123": "user_1", "key-456": "user_2"}

def validate_api_key(x_api_key: str = Header(...)) -> str:
    user = API_KEYS.get(x_api_key)
    if not user:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return user

@app.get("/secure")
async def secure_endpoint(user: str = Depends(validate_api_key)):
    return {"message": f"Hello {user}"}

6. Async APIs

"""API asynchrone avec file d'attente."""

from fastapi import FastAPI, BackgroundTasks, HTTPException
from typing import AsyncIterator
import asyncio

app = FastAPI()

class AsyncTaskManager:
    def __init__(self):
        self.tasks: dict = {}

    async def run_task(self, task_id: str, duration: float) -> None:
        await asyncio.sleep(duration)
        self.tasks[task_id] = {"status": "completed", "result": f"Task done in {duration}s"}

    def create_task(self, duration: float) -> str:
        import uuid
        task_id = str(uuid.uuid4())
        self.tasks[task_id] = {"status": "running"}
        return task_id

manager = AsyncTaskManager()

@app.post("/async-task")
async def start_task(duration: float, background_tasks: BackgroundTasks):
    task_id = manager.create_task(duration)
    background_tasks.add_task(manager.run_task, task_id, duration)
    return {"task_id": task_id, "status": "running"}

@app.get("/async-task/{task_id}")
async def get_task(task_id: str):
    task = manager.tasks.get(task_id)
    if not task:
        raise HTTPException(404, "Task not found")
    return task

7. Tableau Récapitulatif