Middleware pentru API în Python

Cod în jurul fiecărui handler

Middleware este cod care rulează în jurul handlerelor tale: înainte ca cererea să ajungă la ele, după ce răspunsul pleacă de la ele sau în locul lor. Jurnalizarea, autentificarea, limitarea ratei, compresia — toate sunt middleware.

Aceeași idee circulă sub trei nume: framework-urile web spun middleware, gRPC spune interceptor, iar unele ecosisteme spun filter. Un singur concept, trei vocabulare.

În Python, un middleware este o funcție care primește o funcție handler și returnează una împachetată — exact lucrul pentru care a fost creată sintaxa @decorator a limbajului Python și exact tiparul Decorator din grupul despre design patterns. O stivă de astfel de funcții se comportă ca un Chain of Responsibility.

Un middleware de jurnalizare

with_log împachetează orice funcție handler: afișează cererea care sosește, lasă handlerul să lucreze și afișează din nou la ieșire. Handlerul însuși nu știe nimic despre asta. functools.wraps păstrează vizibile numele și documentația funcției împachetate — același ajutor pe care îl folosește orice decorator Python.

Împachetezi o dată, jurnalizezi tot

Handlerul împachetat servește două căi diferite și ambele sunt jurnalizate — jurnalizarea este scrisă o singură dată, în funcția care împachetează, nu în fiecare handler.

Scrie un program Python cu un middleware with_log care împachetează o funcție handler, afișând o linie cu săgeată, metoda și calea înainte de a o apela și o linie de final după aceea. Împachetează o funcție app care servește două căi, pornește serverul pe un port liber, trimite cereri pe ambele căi și afișează răspunsurile.

> Ieșire: program Python complet

Prompt

Scrie un program Python cu un middleware with_log care împachetează o funcție handler, afișând o linie cu săgeată, metoda și calea înainte de a o apela și o linie de final după aceea. Împachetează o funcție app care servește două căi, pornește serverul pe un port liber, trimite cereri pe ambele căi și afișează răspunsurile.

Ieșire: program Python complet

import functools
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

def app(method, path):
  if path == "/users":
    return "Ana, Ben\n"
  if path == "/notes":
    return "shopping, ideas\n"
  return "nimic aici\n"

# Middleware: primește o funcție handler și returnează una împachetată.
def with_log(next_handler):
  @functools.wraps(next_handler)
  def wrapped(method, path):
    print("-> ", method, path)
    body = next_handler(method, path)
    print("<-  gata")
    return body
  return wrapped

handle = with_log(app)

class Handler(BaseHTTPRequestHandler):
  def do_GET(self):
    body = handle("GET", self.path)
    self.send_response(200)
    self.send_header("Content-Type", "text/plain")
    self.end_headers()
    self.wfile.write(body.encode())

  def log_message(self, fmt, *args):
    pass

server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
base = f"http://127.0.0.1:{server.server_address[1]}"
threading.Thread(target=server.serve_forever, daemon=True).start()

for path in ["/users", "/notes"]:
  with urllib.request.urlopen(base + path) as resp:
    print("corp:", resp.read().decode().strip())

server.shutdown()

Middleware care oprește cererea

Un middleware poate și să refuze: o verificare de autentificare care nu găsește o cheie validă răspunde cu 401 Unauthorized și nu mai apelează niciodată handlerul următor.

Înlănțuită ca with_log(with_auth(report)), cererea trece prin straturi în ordine — iar răspunsul se întoarce înapoi prin ele.

Un strat de autentificare în lanț

Cererea fără cheie este oprită de stratul de autentificare cu 401 — handlerul nu rulează deloc. Cu cheia, cererea trece prin tot lanțul și ajunge la date.

Scrie un program Python cu două middleware-uri înlănțuite: with_log, care afișează cererea, și with_auth, care verifică antetul X-API-Key și răspunde cu 401 fără să apeleze handlerul următor atunci când cheia este greșită. Împachetează handlerul ca with_log(with_auth(report)), pornește serverul pe un port liber, apoi trimite o cerere fără cheie și una cu cheie, afișând codurile de stare și corpurile ambelor răspunsuri.

> Ieșire: program Python complet

Prompt

Scrie un program Python cu două middleware-uri înlănțuite: with_log, care afișează cererea, și with_auth, care verifică antetul X-API-Key și răspunde cu 401 fără să apeleze handlerul următor atunci când cheia este greșită. Împachetează handlerul ca with_log(with_auth(report)), pornește serverul pe un port liber, apoi trimite o cerere fără cheie și una cu cheie, afișând codurile de stare și corpurile ambelor răspunsuri.

Ieșire: program Python complet

import threading
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

def report(method, path, headers):
  return 200, "raportul secret\n"

def with_log(next_handler):
  def wrapped(method, path, headers):
    print("->", method, path)
    return next_handler(method, path, headers)
  return wrapped

# Middleware de autentificare: poate opri cererea înainte de handler.
def with_auth(next_handler):
  def wrapped(method, path, headers):
    if headers.get("X-API-Key") != "secret123":
      return 401, "cheie API lipsă sau greșită\n"
    return next_handler(method, path, headers)
  return wrapped

handle = with_log(with_auth(report))

class Handler(BaseHTTPRequestHandler):
  def do_GET(self):
    status, body = handle("GET", self.path, self.headers)
    self.send_response(status)
    self.send_header("Content-Type", "text/plain")
    self.end_headers()
    self.wfile.write(body.encode())

  def log_message(self, fmt, *args):
    pass

server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
base = f"http://127.0.0.1:{server.server_address[1]}"
threading.Thread(target=server.serve_forever, daemon=True).start()

for key in ["", "secret123"]:
  req = urllib.request.Request(base + "/report")
  if key:
    req.add_header("X-API-Key", key)
  try:
    with urllib.request.urlopen(req) as resp:
      status, body = resp.status, resp.read().decode()
  except urllib.error.HTTPError as err:
    status, body = err.code, err.read().decode()
  print("  stare:", status, "-", body.strip())

server.shutdown()