Se cerchi “FastAPI protobuf” troverai molti articoli che cercano di usare i modelli Pydantic come ponte alla serializzazione protobuf. Non esiste tale ponte in Pydantic. L’approccio corretto e piu semplice e diretto: gli endpoint FastAPI possono accettare e restituire bytes grezzi. La libreria Python di protobuf gestisce la codifica effettiva. Questo articolo illustra un’implementazione completa funzionante, la content negotiation e il testing.
Setup#
Installa i pacchetti necessari:
pip install fastapi uvicorn protobuf betterproto grpcio-toolsprotobufe il runtime Python ufficiale Google per le classi generate da.protobetterprotoe un’alternativa che genera dataclass Python invece del vecchio stile imperativogrpcio-toolsincludeprotoce il plugin gRPC per Python, cosi puoi compilare file.protosenza installareprotocseparatamente
Definire e compilare uno schema proto#
syntax = "proto3";
package user;
message Address {
string street = 1;
string city = 2;
string country = 3;
}
message User {
string id = 1;
string name = 2;
string email = 3;
int32 age = 4;
Address address = 5;
}
message CreateUserRequest {
User user = 1;
}Compila in Python usando grpcio-tools (nessun protoc separato necessario):
python -m grpc_tools.protoc \
-I protos \
--python_out=gen \
--grpc_python_out=gen \
protos/user.protoQuesto produce gen/user_pb2.py (tipi dei messaggi) e gen/user_pb2_grpc.py (stub del servizio). Per betterproto, usa --python_betterproto_out=gen al posto di --python_out.
Aggiungi gen/ al .gitignore o committa i file generati: sii coerente nel team. I file generati possono divergere se i membri del team usano versioni diverse di protoc.
Endpoint FastAPI corretto: bytes in ingresso, bytes in uscita#
PROTOBUF non esiste in Pydantic. Il pattern corretto e usare Request direttamente e restituire Response con il content type di protobuf:
from fastapi import FastAPI, Request, Response, HTTPException
from gen.user_pb2 import User, CreateUserRequest
app = FastAPI()
PROTOBUF_CONTENT_TYPE = "application/x-protobuf"
@app.post("/users", response_class=Response)
async def create_user(request: Request) -> Response:
content_type = request.headers.get("content-type", "")
if content_type != PROTOBUF_CONTENT_TYPE:
raise HTTPException(status_code=415, detail="expected application/x-protobuf")
body = await request.body()
try:
req = CreateUserRequest()
req.ParseFromString(body)
except Exception as exc:
raise HTTPException(status_code=400, detail=f"invalid protobuf payload: {exc}")
# Logica di business -- per questo esempio, restituisce l'utente con un ID assegnato.
req.user.id = "usr-generated-001"
return Response(
content=req.user.SerializeToString(),
media_type=PROTOBUF_CONTENT_TYPE,
)ParseFromString modifica l’oggetto in-place e restituisce il numero di byte consumati, non l’oggetto parsato. L’errore comune e scrivere user = User().ParseFromString(body) che assegna un intero, non un User.
Client che invia protobuf#
import requests
from gen.user_pb2 import User, CreateUserRequest
user = User(
name="Alice",
email="alice@example.com",
age=31,
address={"street": "10 Downing Street", "city": "London", "country": "UK"},
)
req = CreateUserRequest(user=user)
response = requests.post(
"http://localhost:8000/users",
data=req.SerializeToString(),
headers={"Content-Type": "application/x-protobuf"},
)
response.raise_for_status()
created = User()
created.ParseFromString(response.content)
print(f"created user id: {created.id}")Content negotiation: JSON e protobuf dallo stesso endpoint#
Puoi servire entrambi i formati da un singolo handler ispezionando l’header Accept. E utile durante una migrazione quando alcuni client hanno gia adottato protobuf e altri no:
import json
from fastapi import FastAPI, Request, Response
from google.protobuf.json_format import MessageToDict
from gen.user_pb2 import User
app = FastAPI()
@app.get("/users/{user_id}", response_class=Response)
async def get_user(user_id: str, request: Request) -> Response:
# Recupera l'utente dallo storage -- semplificato qui.
user = User(
id=user_id,
name="Alice",
email="alice@example.com",
age=31,
)
accept = request.headers.get("accept", "application/json")
if "application/x-protobuf" in accept:
return Response(
content=user.SerializeToString(),
media_type="application/x-protobuf",
)
# Default a JSON usando il serializzatore JSON di protobuf per coerenza dei nomi dei campi.
return Response(
content=json.dumps(MessageToDict(user, preserving_proto_field_name=True)),
media_type="application/json",
)Usa google.protobuf.json_format.MessageToDict con preserving_proto_field_name=True quando converti in JSON. Mantiene i nomi dei campi in snake_case coerenti con la definizione proto, invece di convertirli in camelCase.
Usare betterproto per dataclass piu pulite#
Il protoc --python_out predefinito genera un’API imperativa verbosa. betterproto genera dataclass Python piu idiomatiche:
# Con il codice generato da betterproto, l'API diventa stile dataclass.
# Compila con: python -m grpc_tools.protoc ... --python_betterproto_out=gen_bt
import betterproto
from gen_bt.user import User, Address, CreateUserRequest
async def handle_create(body: bytes) -> bytes:
req = CreateUserRequest().parse(body)
req.user.id = "usr-generated-001"
return bytes(req.user) # gli oggetti betterproto si serializzano con bytes()
# In un handler FastAPI:
# body = await request.body()
# result_bytes = await handle_create(body)
# return Response(content=result_bytes, media_type="application/x-protobuf")from gen.user_pb2 import User
# Parse
user = User()
user.ParseFromString(raw_bytes)
# Serialize
raw_bytes = user.SerializeToString()Pro: supporto ufficiale Google, stabile, ampiamente compatibile. Contro: codice generato verboso, non stile dataclass.
from gen_bt.user import User
# Parse
user = User().parse(raw_bytes)
# Serialize
raw_bytes = bytes(user)Pro: dataclass pulite, type hint, async-friendly, nessuna magia descriptor pool. Contro: terze parti, alcuni casi limite di proto3 differiscono dall’implementazione di riferimento.
Testing degli endpoint protobuf con pytest#
import pytest
from fastapi.testclient import TestClient
from gen.user_pb2 import User, CreateUserRequest
from app.main import app
client = TestClient(app)
def test_create_user_protobuf():
user = User(name="Alice", email="alice@example.com", age=31)
req = CreateUserRequest(user=user)
response = client.post(
"/users",
content=req.SerializeToString(),
headers={"content-type": "application/x-protobuf"},
)
assert response.status_code == 200
assert response.headers["content-type"] == "application/x-protobuf"
created = User()
created.ParseFromString(response.content)
assert created.name == "Alice"
assert created.id != ""
def test_create_user_wrong_content_type():
response = client.post(
"/users",
content=b"{}",
headers={"content-type": "application/json"},
)
assert response.status_code == 415Alternativa: pattern gRPC-Gateway#
Se hai bisogno sia di un’interfaccia gRPC che di un’interfaccia REST/JSON dallo stesso servizio, il pattern gRPC-Gateway vale la pena di essere conosciuto. Invece di due applicazioni separate, definisci le annotazioni HTTP nel file .proto e un proxy gateway traduce le chiamate REST in gRPC:
flowchart LR
Browser["Browser / curl\n(JSON + REST)"] --> GW["gRPC-Gateway\n(proxy grpc-gateway)"]
GRPCClient["gRPC Client\n(Go / Python / Java)"] --> SVC
GW -->|transcodifica JSON in protobuf| SVC["gRPC Service\n(il tuo server Go)"]
Per i servizi Python questo e tipicamente gestito eseguendo grpc-gateway come sidecar o usando grpcio direttamente accanto a FastAPI. L’approccio FastAPI in questo articolo e piu semplice e appropriato quando possiedi tutti i client e vuoi un singolo endpoint binario.
Se vuoi approfondire questi argomenti, offro sessioni di coaching 1:1 per ingegneri che lavorano su integrazione AI, architettura cloud e platform engineering. Prenota una sessione (50 EUR / 60 min) o scrivimi a manuel.fedele+website@gmail.com.
