Web Python — Cours Complet

1. FastAPI — APIs Modernes et Rapides

Installation

pip install fastapi uvicorn[standard]
pip install sqlalchemy asyncpg pytest httpx

Structure d'un Projet FastAPI

app/
├── __init__.py
├── main.py            # Point d'entrée
├── routers/           # Routes
│   ├── __init__.py
│   ├── items.py
│   └── users.py
├── models/            # SQLAlchemy
│   ├── __init__.py
│   └── models.py
├── schemas/           # Pydantic
│   ├── __init__.py
│   └── schemas.py
├── dependencies/      # Dépendances injectables
│   ├── __init__.py
│   └── auth.py
└── services/          # Logique métier
    ├── __init__.py
    └── user_service.py

Application de Base

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print("Starting up...")
    yield
    # Shutdown
    print("Shutting down...")

app = FastAPI(
    title="Modern API",
    version="0.1.0",
    lifespan=lifespan,
)

@app.get("/")
async def root() -> dict:
    return {"message": "Hello World"}

@app.get("/health")
async def health() -> dict:
    return {"status": "healthy"}

Routers — Organisation des Routes

# routers/items.py
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel
from typing import Optional

router = APIRouter(prefix="/items", tags=["items"])

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

class ItemResponse(Item):
    id: int

ITEMS_DB: dict[int, Item] = {}

@router.post("/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item) -> ItemResponse:
    item_id = len(ITEMS_DB) + 1
    ITEMS_DB[item_id] = item
    return ItemResponse(id=item_id, **item.model_dump())

@router.get("/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int) -> ItemResponse:
    if item_id not in ITEMS_DB:
        raise HTTPException(status_code=404, detail="Item not found")
    return ItemResponse(id=item_id, **ITEMS_DB[item_id].model_dump())

@router.get("/", response_model=list[ItemResponse])
async def list_items(skip: int = 0, limit: int = 10) -> list[ItemResponse]:
    items = []
    for iid in list(ITEMS_DB.keys())[skip:skip + limit]:
        items.append(ItemResponse(id=iid, **ITEMS_DB[iid].model_dump()))
    return items

@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int) -> None:
    if item_id not in ITEMS_DB:
        raise HTTPException(status_code=404, detail="Item not found")
    del ITEMS_DB[item_id]

Dépendances — Injection de Dépendances

# dependencies/auth.py
from fastapi import Depends, HTTPException, Header
from typing import Annotated

async def verify_token(authorization: str = Header(...)) -> dict:
    """Vérifie le token JWT et retourne l'utilisateur."""
    # Logique de vérification
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"user_id": 1, "role": "admin"}

# Dépendance réutilisable
CurrentUser = Annotated[dict, Depends(verify_token)]

# Usage
@router.get("/protected")
async def protected_route(user: CurrentUser) -> dict:
    return {"message": f"Hello {user['user_id']}"}

# Dépendance avec paramètres
from fastapi import Query

def pagination(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
) -> dict:
    return {"skip": skip, "limit": limit}

Pagination = Annotated[dict, Depends(pagination)]

@router.get("/paginated")
async def list_paginated(p: Pagination) -> dict:
    return {"skip": p["skip"], "limit": p["limit"]}

Validation Pydantic

from pydantic import BaseModel, Field, EmailStr, ConfigDict
from typing import Optional
from datetime import datetime

class UserCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")  # pas de champs supplémentaires

    username: str = Field(..., min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$")
    email: EmailStr
    age: int = Field(ge=0, le=150)
    password: str = Field(..., min_length=8)

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    created_at: datetime

Background Tasks

from fastapi import BackgroundTasks

def send_email(email: str, message: str) -> None:
    """Envoie un email (synchrone)."""
    print(f"Sending email to {email}: {message}")

@router.post("/register")
async def register(
    user: UserCreate,
    background_tasks: BackgroundTasks,
) -> dict:
    user_id = 1  # créer l'utilisateur
    background_tasks.add_task(
        send_email, user.email, "Welcome!"
    )
    return {"user_id": user_id, "status": "created"}

Middleware

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        elapsed = time.perf_counter() - start
        response.headers["X-Process-Time"] = str(elapsed)
        return response

# Ajout dans main.py
app.add_middleware(TimingMiddleware)

# CORS
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

WebSocket

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active_connections.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active_connections.remove(ws)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@router.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            data = await ws.receive_text()
            await manager.broadcast(f"Message: {data}")
    except WebSocketDisconnect:
        manager.disconnect(ws)

Testing FastAPI

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "Test", "price": 10.0},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Test"
    assert data["id"] is not None

2. Django — Framework Complet

Structure Django

mysite/
├── manage.py
├── mysite/
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── myapp/
    ├── models.py
    ├── views.py
    ├── serializers.py
    ├── urls.py
    └── templates/

Models

# myapp/models.py
from django.db import models

class Category(models.Model):
    name = models.CharField(max_length=100, unique=True)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        verbose_name_plural = "categories"

    def __str__(self) -> str:
        return self.name

class Product(models.Model):
    name = models.CharField(max_length=200)
    category = models.ForeignKey(
        Category, on_delete=models.CASCADE, related_name="products"
    )
    price = models.DecimalField(max_digits=10, decimal_places=2)
    stock = models.IntegerField(default=0)
    active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self) -> str:
        return f"{self.name} ({self.price}€)"

Views (CBV)

# myapp/views.py
from django.views.generic import ListView, DetailView, CreateView
from django.urls import reverse_lazy
from .models import Product

class ProductListView(ListView):
    model = Product
    template_name = "products/list.html"
    context_object_name = "products"
    paginate_by = 20

class ProductDetailView(DetailView):
    model = Product
    template_name = "products/detail.html"

class ProductCreateView(CreateView):
    model = Product
    fields = ["name", "category", "price", "stock"]
    template_name = "products/form.html"
    success_url = reverse_lazy("product-list")

Django REST Framework

# serializers.py
from rest_framework import serializers
from .models import Product, Category

class CategorySerializer(serializers.ModelSerializer):
    class Meta:
        model = Category
        fields = ["id", "name"]

class ProductSerializer(serializers.ModelSerializer):
    category = CategorySerializer(read_only=True)
    category_id = serializers.IntegerField(write_only=True)

    class Meta:
        model = Product
        fields = ["id", "name", "category", "category_id", "price", "stock"]

# views.py
from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated

class ProductViewSet(viewsets.ModelViewSet):
    queryset = Product.objects.select_related("category").all()
    serializer_class = ProductSerializer
    permission_classes = [IsAuthenticated]

3. Comparaison des Frameworks

WSGI vs ASGI

Diagramme en cours de génération...

4. Template Engines (Jinja2)

from jinja2 import Environment, FileSystemLoader

env = Environment(
    loader=FileSystemLoader("templates"),
    autoescape=True,
)

template = env.get_template("hello.html")
html = template.render(name="World", items=[1, 2, 3])

<!-- templates/hello.html -->
<!DOCTYPE html>
<html>
<head><title>Hello {{ name }}</title></head>
<body>
    <h1>Hello, {{ name }}!</h1>
    <ul>
    {% for item in items %}
        <li>{{ item }}</li>
    {% endfor %}
    </ul>
</body>
</html>

5. ORM Patterns

SQLAlchemy (async)

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import select
from typing import Optional

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    email: Mapped[str] = mapped_column(unique=True)

engine = create_async_engine("postgresql+asyncpg://localhost/db")

async def get_user(name: str) -> Optional[User]:
    async with AsyncSession(engine) as session:
        result = await session.execute(
            select(User).where(User.name == name)
        )
        return result.scalar_one_or_none()

Django ORM

# Queries avancées
from django.db.models import Q, Count, Sum, Avg

# Filtres complexes
products = Product.objects.filter(
    Q(price__gt=100) | Q(stock=0),
    active=True,
)

# Agrégations
stats = Product.objects.aggregate(
    avg_price=Avg("price"),
    total_stock=Sum("stock"),
    count=Count("id"),
)

# Optimisation (N+1)
products = Product.objects.select_related("category").all()
products = Product.objects.prefetch_related("tags").all()

6. Tableau Récapitulatif