Module 06: Project — Build a gRPC Microservice System

What you're building

A working three-part order-processing system, built step by step. Unlike the open-ended capstones elsewhere in backend/, this one gives you the code — type it in, run it, and watch each of modules 00-05 do its job in a real system rather than an isolated exercise.

                      JWT (end user)
   HTTP/JSON  ┌──────────────────┐
  ──────────▶ │  gateway         │   FastAPI — the only public surface
              │  (REST facade)   │
              └────────┬─────────┘
                       │ gRPC + mTLS + propagated user token
                       ▼
              ┌──────────────────┐        gRPC + mTLS
              │  orders          │ ─────────────────────┐
              │  - CreateOrder   │                      ▼
              │  - WatchOrders   │            ┌──────────────────┐
              │    (srv stream)  │            │  inventory       │
              └──────────────────┘            │  - CheckStock    │
                                              │  - BulkRestock   │
                                              │    (cli stream)  │
                                              └──────────────────┘

By the end you'll have exercised: versioned protobuf schemas with safe evolution (00), mTLS between services plus per-user JWTs (01), health checking, graceful shutdown and client-side load balancing (02), logging and metrics interceptors plus a real test suite (03), a REST gateway (04), and a ghz benchmark (05).

Budget 6-10 hours. Do it in order; each step runs before you move on.

Setup

mkdir grpc-shop && cd grpc-shop
python -m venv venv && source venv/bin/activate    # Windows: venv\Scripts\activate
pip install grpcio grpcio-tools grpcio-health-checking grpcio-reflection \
            protobuf pyjwt prometheus-client fastapi uvicorn httpx pytest

Optional but used later: grpcurl, ghz, Docker.

Step 1 — Repository layout and the schemas

grpc-shop/
  proto/shop/v1/{common.proto,inventory.proto,orders.proto}
  gen/                     # generated code (git-ignored)
  services/{inventory,orders,gateway}
  shared/{interceptors.py,serving.py,security.py}
  certs/
  tests/

proto/shop/v1/common.proto:

syntax = "proto3";
package shop.v1;

import "google/protobuf/timestamp.proto";

enum OrderStatus {
  ORDER_STATUS_UNSPECIFIED = 0;   // module 00: zero value is always "unset"
  ORDER_STATUS_PENDING     = 1;
  ORDER_STATUS_CONFIRMED   = 2;
  ORDER_STATUS_REJECTED    = 3;
}

message LineItem {
  string sku = 1;
  int32  quantity = 2;
}

message Order {
  reserved 4;                     // module 00: was `int64 total_cents`, replaced by 7
  reserved "total_cents";

  string id = 1;
  string customer_id = 2;
  repeated LineItem items = 3;
  OrderStatus status = 5;
  google.protobuf.Timestamp created_at = 6;
  double total = 7;
  optional string rejection_reason = 8;   // explicit presence: absent != ""
}

proto/shop/v1/inventory.proto:

syntax = "proto3";
package shop.v1;

import "shop/v1/common.proto";

service InventoryService {
  rpc CheckStock(CheckStockRequest) returns (CheckStockResponse);
  rpc BulkRestock(stream RestockItem) returns (BulkRestockSummary);   // client streaming
}

message CheckStockRequest  { repeated LineItem items = 1; }
message StockResult        { string sku = 1; int32 available = 2; bool sufficient = 3; double unit_price = 4; }
message CheckStockResponse { repeated StockResult results = 1; bool all_available = 2; }

message RestockItem        { string sku = 1; int32 quantity = 2; }
message BulkRestockSummary { int32 items_processed = 1; int32 skus_updated = 2; }

proto/shop/v1/orders.proto:

syntax = "proto3";
package shop.v1;

import "shop/v1/common.proto";

service OrderService {
  rpc CreateOrder(CreateOrderRequest) returns (CreateOrderResponse);
  rpc GetOrder(GetOrderRequest) returns (GetOrderResponse);
  rpc WatchOrders(WatchOrdersRequest) returns (stream Order);        // server streaming
}

message CreateOrderRequest  { string customer_id = 1; repeated LineItem items = 2; }
message CreateOrderResponse { Order order = 1; }
message GetOrderRequest     { string id = 1; }
message GetOrderResponse    { Order order = 1; }
message WatchOrdersRequest  { string customer_id = 1; int32 limit = 2; }

Note what's already in play: a versioned package (shop.v1), a reserved field from a past migration, an UNSPECIFIED enum zero, optional for a field where absent and empty differ, and both streaming directions.

Step 2 — Code generation

scripts/gen.sh:

#!/usr/bin/env bash
set -euo pipefail
rm -rf gen && mkdir -p gen
python -m grpc_tools.protoc -I proto \
  --python_out=gen --grpc_python_out=gen --pyi_out=gen \
  proto/shop/v1/*.proto
find gen -type d -exec touch {}/__init__.py \;
echo "generated:"; find gen -name "*_pb2*.py" | sort

chmod +x scripts/gen.sh && ./scripts/gen.sh

Generated gRPC modules import each other as from shop.v1 import ..., so gen/ must be on sys.path. Create conftest.py at the repo root:

import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "gen"))

and for the services, a shared/bootstrap.py imported first:

import sys, os
_GEN = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "gen")
if _GEN not in sys.path:
    sys.path.insert(0, _GEN)

Step 3 — The inventory service

services/inventory/service.py:

import shared.bootstrap  # noqa: F401  — must come first, puts gen/ on sys.path
import grpc
from shop.v1 import inventory_pb2, inventory_pb2_grpc

STOCK = {"SKU-1": (100, 9.99), "SKU-2": (5, 24.50), "SKU-3": (0, 5.00)}


class InventoryService(inventory_pb2_grpc.InventoryServiceServicer):
    def __init__(self):
        self._stock = dict(STOCK)

    def CheckStock(self, request, context):
        if not request.items:
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "items must not be empty")

        results, all_ok = [], True
        for item in request.items:
            if item.quantity <= 0:
                context.abort(grpc.StatusCode.INVALID_ARGUMENT,
                              f"quantity for {item.sku} must be positive")
            available, price = self._stock.get(item.sku, (0, 0.0))
            sufficient = available >= item.quantity
            all_ok &= sufficient
            results.append(inventory_pb2.StockResult(
                sku=item.sku, available=available,
                sufficient=sufficient, unit_price=price,
            ))
        return inventory_pb2.CheckStockResponse(results=results, all_available=all_ok)

    def BulkRestock(self, request_iterator, context):
        # client streaming: consume the whole stream, then return one summary
        processed, touched = 0, set()
        for item in request_iterator:
            if item.quantity <= 0:
                context.abort(grpc.StatusCode.INVALID_ARGUMENT, "quantity must be positive")
            available, price = self._stock.get(item.sku, (0, 0.0))
            self._stock[item.sku] = (available + item.quantity, price)
            processed += 1
            touched.add(item.sku)
        return inventory_pb2.BulkRestockSummary(
            items_processed=processed, skus_updated=len(touched))

Step 4 — Shared serving helpers

This is where modules 02 and 03 become reusable infrastructure rather than copy-paste. shared/serving.py:

import time, signal, logging
from concurrent import futures
import grpc
from grpc_health.v1 import health, health_pb2, health_pb2_grpc
from grpc_reflection.v1alpha import reflection

log = logging.getLogger("serving")

SERVER_OPTIONS = [
    ("grpc.keepalive_time_ms", 30000),
    ("grpc.keepalive_timeout_ms", 10000),
    ("grpc.keepalive_permit_without_calls", 1),
    # must tolerate clients pinging as often as they're configured to (module 02)
    ("grpc.http2.min_ping_interval_without_data_ms", 10000),
    ("grpc.http2.max_ping_strikes", 0),
    ("grpc.max_receive_message_length", 16 * 1024 * 1024),
    ("grpc.max_connection_age_ms", 300000),
    ("grpc.max_connection_age_grace_ms", 30000),
]

CLIENT_OPTIONS = [
    ("grpc.keepalive_time_ms", 30000),
    ("grpc.keepalive_timeout_ms", 10000),
    ("grpc.keepalive_permit_without_calls", 1),
    # module 02: without this, gRPC uses pick_first and one backend takes everything
    ("grpc.service_config", '{"loadBalancingConfig": [{"round_robin": {}}]}'),
]


def build_server(register_fn, service_names, interceptors=(), max_workers=16, creds=None, port=50051):
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=max_workers),
        interceptors=list(interceptors),
        options=SERVER_OPTIONS,
    )
    register_fn(server)

    health_servicer = health.HealthServicer()
    health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)

    reflection.enable_server_reflection(
        tuple(service_names) + (health_pb2.DESCRIPTOR.services_by_name["Health"].full_name,
                                reflection.SERVICE_NAME),
        server,
    )

    addr = f"[::]:{port}"
    if creds is not None:
        server.add_secure_port(addr, creds)
    else:
        server.add_insecure_port(addr)
    return server, health_servicer


def serve_forever(server, health_servicer, names):
    server.start()
    for n in ("",) + tuple(names):
        health_servicer.set(n, health_pb2.HealthCheckResponse.SERVING)
    log.info("serving %s", names)

    def shutdown(signum, frame):
        # module 02 ordering: fail readiness -> wait for the LB -> drain -> exit
        for n in ("",) + tuple(names):
            health_servicer.set(n, health_pb2.HealthCheckResponse.NOT_SERVING)
        log.info("draining…")
        time.sleep(3)
        server.stop(grace=30).wait()

    signal.signal(signal.SIGTERM, shutdown)
    signal.signal(signal.SIGINT, shutdown)
    server.wait_for_termination()

Step 5 — Interceptors: auth, logging, metrics

shared/interceptors.py. The handler-kind dispatch from module 03 matters here — this system has streaming RPCs in both directions, so a unary-only interceptor would leave them uninstrumented and unauthenticated.

import time, logging
import grpc
from prometheus_client import Counter, Histogram

log = logging.getLogger("rpc")

RPC_STARTED = Counter("grpc_server_started_total", "RPCs started", ["method"])
RPC_HANDLED = Counter("grpc_server_handled_total", "RPCs completed", ["method", "code"])
RPC_LATENCY = Histogram("grpc_server_handling_seconds", "RPC latency", ["method"],
                        buckets=(.001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10))

_HANDLER_FACTORY = {
    "unary_unary": grpc.unary_unary_rpc_method_handler,
    "unary_stream": grpc.unary_stream_rpc_method_handler,
    "stream_unary": grpc.stream_unary_rpc_method_handler,
    "stream_stream": grpc.stream_stream_rpc_method_handler,
}


def _kind(handler):
    for k in _HANDLER_FACTORY:
        if getattr(handler, k) is not None:
            return k
    raise RuntimeError("unrecognised handler")


def _rewrap(handler, kind, fn):
    # module 03: dropping these serializers is the classic interceptor bug
    return _HANDLER_FACTORY[kind](
        fn,
        request_deserializer=handler.request_deserializer,
        response_serializer=handler.response_serializer,
    )


class ObservabilityInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, details):
        handler = continuation(details)
        if handler is None:
            return None
        kind = _kind(handler)
        inner = getattr(handler, kind)
        method = details.method

        def wrapper(request, context):
            RPC_STARTED.labels(method=method).inc()
            start = time.perf_counter()

            def finish(streamed=None):
                code = context.code()
                RPC_LATENCY.labels(method=method).observe(time.perf_counter() - start)
                RPC_HANDLED.labels(method=method, code=(code.name if code else "OK")).inc()
                log.info("rpc %s code=%s dur_ms=%.2f%s", method,
                         code.name if code else "OK",
                         (time.perf_counter() - start) * 1000,
                         "" if streamed is None else f" msgs={streamed}")

            try:
                result = inner(request, context)
                if kind in ("unary_stream", "stream_stream"):
                    def counted():
                        n = 0
                        try:
                            for item in result:
                                n += 1
                                yield item
                        finally:
                            finish(n)      # module 03: time the *consumption*
                    return counted()
                finish()
                return result
            except Exception:
                finish()
                raise

        return _rewrap(handler, kind, wrapper)


class AuthInterceptor(grpc.ServerInterceptor):
    """Fails closed. Exempts health/reflection so probes and tooling work."""

    EXEMPT_PREFIXES = ("/grpc.health.v1.Health/", "/grpc.reflection.")

    def __init__(self, verify):
        self._verify = verify

    def intercept_service(self, continuation, details):
        if details.method.startswith(self.EXEMPT_PREFIXES):
            return continuation(details)

        md = dict(details.invocation_metadata or ())
        token = md.get("authorization", "")        # lowercase: gRPC normalises keys
        if not token.startswith("Bearer ") or not self._verify(token[7:]):
            handler = continuation(details)
            kind = _kind(handler) if handler else "unary_unary"

            def deny(request, context):
                context.abort(grpc.StatusCode.UNAUTHENTICATED, "invalid or missing token")

            # must match the real handler's kind, or the client sees a confusing
            # "unary response for streaming call" error instead of UNAUTHENTICATED
            return _rewrap(handler, kind, deny) if handler else \
                grpc.unary_unary_rpc_method_handler(deny)
        return continuation(details)

Step 6 — Security: JWTs and mTLS

shared/security.py:

import os
import datetime as dt
import jwt

# HS256 needs >= 32 bytes; newer PyJWT warns loudly below that.
# Load from the environment in anything real — this default is for local dev only.
SECRET = os.getenv("JWT_SECRET", "dev-only-secret-change-me-32-bytes-min")
ALGO = "HS256"


def issue(sub: str, minutes: int = 30) -> str:
    now = dt.datetime.now(dt.timezone.utc)
    return jwt.encode({"sub": sub, "iat": now,
                       "exp": now + dt.timedelta(minutes=minutes)}, SECRET, algorithm=ALGO)


def verify(token: str) -> bool:
    try:
        jwt.decode(token, SECRET, algorithms=[ALGO])
        return True
    except jwt.PyJWTError:
        return False


def subject(token: str) -> str | None:
    try:
        return jwt.decode(token, SECRET, algorithms=[ALGO]).get("sub")
    except jwt.PyJWTError:
        return None

Certificates (module 01) — scripts/certs.sh:

#!/usr/bin/env bash
set -euo pipefail
mkdir -p certs && cd certs

openssl req -x509 -newkey rsa:2048 -nodes -days 365 \
  -keyout ca.key -out ca.crt -subj "/CN=shop-dev-ca" 2>/dev/null

issue() {   # $1 = name, $2 = SAN
  openssl req -newkey rsa:2048 -nodes -keyout "$1.key" -out "$1.csr" \
    -subj "/CN=$1" 2>/dev/null
  openssl x509 -req -in "$1.csr" -CA ca.crt -CAkey ca.key -CAcreateserial \
    -out "$1.crt" -days 365 -extfile <(printf "subjectAltName=%s" "$2") 2>/dev/null
  rm "$1.csr"
}

issue inventory "DNS:localhost,DNS:inventory,IP:127.0.0.1"
issue orders    "DNS:localhost,DNS:orders,IP:127.0.0.1"
issue gateway   "DNS:localhost,DNS:gateway,IP:127.0.0.1"
echo "certs written to $(pwd)"

# shared/security.py (continued)
import grpc, pathlib

CERTS = pathlib.Path(__file__).resolve().parent.parent / "certs"


def _read(name):
    return (CERTS / name).read_bytes()


def server_creds(name):
    return grpc.ssl_server_credentials(
        [(_read(f"{name}.key"), _read(f"{name}.crt"))],   # key FIRST (module 01)
        root_certificates=_read("ca.crt"),
        require_client_auth=True,                          # this is what makes it mTLS
    )


def client_creds(name):
    return grpc.ssl_channel_credentials(
        root_certificates=_read("ca.crt"),
        private_key=_read(f"{name}.key"),
        certificate_chain=_read(f"{name}.crt"),
    )

Step 7 — Running the inventory server

services/inventory/main.py:

import shared.bootstrap  # noqa: F401
import logging, os
from prometheus_client import start_http_server
from shop.v1 import inventory_pb2, inventory_pb2_grpc
from services.inventory.service import InventoryService
from shared.serving import build_server, serve_forever
from shared.interceptors import ObservabilityInterceptor, AuthInterceptor
from shared import security

NAME = inventory_pb2.DESCRIPTOR.services_by_name["InventoryService"].full_name

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(message)s")
    start_http_server(9101)

    use_tls = os.getenv("TLS", "1") == "1"
    server, hs = build_server(
        register_fn=lambda s: inventory_pb2_grpc.add_InventoryServiceServicer_to_server(
            InventoryService(), s),
        service_names=[NAME],
        interceptors=[ObservabilityInterceptor(), AuthInterceptor(security.verify)],
        creds=security.server_creds("inventory") if use_tls else None,
        port=50052,
    )
    serve_forever(server, hs, [NAME])

./scripts/certs.sh
PYTHONPATH=. TLS=0 python services/inventory/main.py

Verify it (auth is on, so an unauthenticated call must fail):

grpcurl -plaintext localhost:50052 list
grpcurl -plaintext localhost:50052 grpc.health.v1.Health/Check
grpcurl -plaintext -d '{"items":[{"sku":"SKU-1","quantity":2}]}' \
        localhost:50052 shop.v1.InventoryService/CheckStock       # expect UNAUTHENTICATED
TOKEN=$(PYTHONPATH=. python -c "from shared import security; print(security.issue('u1'))")
grpcurl -plaintext -H "authorization: Bearer $TOKEN" \
        -d '{"items":[{"sku":"SKU-1","quantity":2}]}' \
        localhost:50052 shop.v1.InventoryService/CheckStock

Expected: list and Health/Check work without a token (they're exempt), the unauthenticated CheckStock returns UNAUTHENTICATED, and the authenticated one returns stock results.

Step 8 — The orders service (a gRPC client and server)

services/orders/service.py:

import shared.bootstrap  # noqa: F401
import uuid, datetime as dt, threading
import grpc
from google.protobuf.timestamp_pb2 import Timestamp
from shop.v1 import common_pb2, orders_pb2, orders_pb2_grpc, inventory_pb2, inventory_pb2_grpc


class OrderService(orders_pb2_grpc.OrderServiceServicer):
    def __init__(self, inventory_channel):
        self._inv = inventory_pb2_grpc.InventoryServiceStub(inventory_channel)
        self._orders = {}
        self._lock = threading.Lock()

    @staticmethod
    def _caller_token(context):
        return dict(context.invocation_metadata()).get("authorization")

    def CreateOrder(self, request, context):
        if not request.items:
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "order must contain items")

        # Propagate the caller's credentials downstream, and bound the call with a
        # deadline so a slow inventory service can't hang this one indefinitely.
        md = [("authorization", self._caller_token(context) or "")]
        try:
            stock = self._inv.CheckStock(
                inventory_pb2.CheckStockRequest(items=request.items),
                metadata=md, timeout=2.0,
            )
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
                context.abort(grpc.StatusCode.UNAVAILABLE, "inventory timed out")
            if e.code() == grpc.StatusCode.INVALID_ARGUMENT:
                context.abort(grpc.StatusCode.INVALID_ARGUMENT, e.details())
            context.abort(grpc.StatusCode.UNAVAILABLE, f"inventory unavailable: {e.code().name}")

        ts = Timestamp(); ts.FromDatetime(dt.datetime.now(dt.timezone.utc))
        order = common_pb2.Order(
            id=f"ORD-{uuid.uuid4().hex[:8]}",
            customer_id=request.customer_id,
            items=list(request.items),
            created_at=ts,
        )

        if stock.all_available:
            order.status = common_pb2.ORDER_STATUS_CONFIRMED
            order.total = sum(r.unit_price * i.quantity
                              for r, i in zip(stock.results, request.items))
        else:
            order.status = common_pb2.ORDER_STATUS_REJECTED
            short = [r.sku for r in stock.results if not r.sufficient]
            order.rejection_reason = "insufficient stock: " + ", ".join(short)

        with self._lock:
            self._orders[order.id] = order
        return orders_pb2.CreateOrderResponse(order=order)

    def GetOrder(self, request, context):
        with self._lock:
            order = self._orders.get(request.id)
        if order is None:
            context.abort(grpc.StatusCode.NOT_FOUND, f"order {request.id} not found")
        return orders_pb2.GetOrderResponse(order=order)

    def WatchOrders(self, request, context):
        # server streaming: yield matching orders, stopping if the client goes away
        with self._lock:
            matching = [o for o in self._orders.values()
                        if o.customer_id == request.customer_id]
        limit = request.limit or len(matching)
        for order in matching[:limit]:
            if not context.is_active():      # client cancelled or deadline expired
                return
            yield order

services/orders/main.py:

import shared.bootstrap  # noqa: F401
import logging, os
import grpc
from prometheus_client import start_http_server
from shop.v1 import orders_pb2, orders_pb2_grpc
from services.orders.service import OrderService
from shared.serving import build_server, serve_forever, CLIENT_OPTIONS
from shared.interceptors import ObservabilityInterceptor, AuthInterceptor
from shared import security

NAME = orders_pb2.DESCRIPTOR.services_by_name["OrderService"].full_name

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(message)s")
    start_http_server(9102)

    use_tls = os.getenv("TLS", "1") == "1"
    target = os.getenv("INVENTORY_ADDR", "localhost:50052")

    if use_tls:
        inv_channel = grpc.secure_channel(target, security.client_creds("orders"),
                                          options=CLIENT_OPTIONS)
    else:
        inv_channel = grpc.insecure_channel(target, options=CLIENT_OPTIONS)

    server, hs = build_server(
        register_fn=lambda s: orders_pb2_grpc.add_OrderServiceServicer_to_server(
            OrderService(inv_channel), s),
        service_names=[NAME],
        interceptors=[ObservabilityInterceptor(), AuthInterceptor(security.verify)],
        creds=security.server_creds("orders") if use_tls else None,
        port=50051,
    )
    serve_forever(server, hs, [NAME])

PYTHONPATH=. TLS=0 python services/orders/main.py
grpcurl -plaintext -H "authorization: Bearer $TOKEN" \
  -d '{"customer_id":"C1","items":[{"sku":"SKU-1","quantity":2}]}' \
  localhost:50051 shop.v1.OrderService/CreateOrder

Expected: a confirmed order with a computed total. Now order SKU-3 (stock 0) and confirm you get ORDER_STATUS_REJECTED with a rejection_reason — the optional field from step 1 doing its job.

Step 9 — The REST gateway

services/gateway/main.py (module 04's "both protocols, shared logic" pattern, with gRPC as the logic layer):

import shared.bootstrap  # noqa: F401
import os
import grpc
from fastapi import FastAPI, HTTPException, Header
from google.protobuf.json_format import MessageToDict
from shop.v1 import orders_pb2, orders_pb2_grpc, common_pb2
from shared.serving import CLIENT_OPTIONS
from shared import security

app = FastAPI(title="Shop Gateway")

_target = os.getenv("ORDERS_ADDR", "localhost:50051")
_channel = (grpc.secure_channel(_target, security.client_creds("gateway"), options=CLIENT_OPTIONS)
            if os.getenv("TLS", "1") == "1"
            else grpc.insecure_channel(_target, options=CLIENT_OPTIONS))
_stub = orders_pb2_grpc.OrderServiceStub(_channel)

# module 04: the standard gRPC -> HTTP status mapping
STATUS = {
    grpc.StatusCode.INVALID_ARGUMENT: 400,
    grpc.StatusCode.UNAUTHENTICATED: 401,
    grpc.StatusCode.PERMISSION_DENIED: 403,
    grpc.StatusCode.NOT_FOUND: 404,
    grpc.StatusCode.ALREADY_EXISTS: 409,
    grpc.StatusCode.RESOURCE_EXHAUSTED: 429,
    grpc.StatusCode.UNIMPLEMENTED: 501,
    grpc.StatusCode.UNAVAILABLE: 503,
    grpc.StatusCode.DEADLINE_EXCEEDED: 504,
}


def _md(authorization: str | None):
    if not authorization:
        raise HTTPException(401, "missing Authorization header")
    return [("authorization", authorization)]


def _call(fn, *args, **kwargs):
    try:
        return fn(*args, **kwargs)
    except grpc.RpcError as e:
        raise HTTPException(STATUS.get(e.code(), 500), e.details() or e.code().name)


def _json(msg):
    # always_print_fields_with_no_presence: keep zero/empty fields in the JSON so
    # REST clients get a stable shape (module 04, exercise 8)
    return MessageToDict(msg, preserving_proto_field_name=True,
                         always_print_fields_with_no_presence=True)


@app.post("/v1/orders", status_code=201)
def create_order(body: dict, authorization: str | None = Header(default=None)):
    req = orders_pb2.CreateOrderRequest(
        customer_id=body.get("customer_id", ""),
        items=[common_pb2.LineItem(sku=i["sku"], quantity=i["quantity"])
               for i in body.get("items", [])],
    )
    return _json(_call(_stub.CreateOrder, req, metadata=_md(authorization), timeout=5).order)


@app.get("/v1/orders/{order_id}")
def get_order(order_id: str, authorization: str | None = Header(default=None)):
    req = orders_pb2.GetOrderRequest(id=order_id)
    return _json(_call(_stub.GetOrder, req, metadata=_md(authorization), timeout=5).order)


@app.get("/v1/customers/{customer_id}/orders")
def watch(customer_id: str, limit: int = 10, authorization: str | None = Header(default=None)):
    req = orders_pb2.WatchOrdersRequest(customer_id=customer_id, limit=limit)
    stream = _call(_stub.WatchOrders, req, metadata=_md(authorization), timeout=10)
    return [_json(o) for o in stream]


@app.get("/healthz")
def healthz():
    return {"status": "ok"}

PYTHONPATH=. TLS=0 uvicorn services.gateway.main:app --port 8000
curl -s -X POST localhost:8000/v1/orders -H "Authorization: Bearer $TOKEN" \
     -H 'content-type: application/json' \
     -d '{"customer_id":"C1","items":[{"sku":"SKU-1","quantity":2}]}' | python -m json.tool
curl -si localhost:8000/v1/orders/NOPE -H "Authorization: Bearer $TOKEN" | head -1

Expected: a JSON order on the first call, and HTTP/1.1 404 on the second — the status mapping working end to end from NOT_FOUND.

Step 10 — Tests

tests/conftest.py:

import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "gen"))
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import pytest, grpc
from concurrent import futures
from shop.v1 import inventory_pb2_grpc, orders_pb2_grpc
from services.inventory.service import InventoryService
from services.orders.service import OrderService
from shared.interceptors import AuthInterceptor
from shared import security


def _serve(register, interceptors=()):
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=4),
                         interceptors=list(interceptors))
    register(server)
    port = server.add_insecure_port("localhost:0")   # module 03: ephemeral port
    server.start()
    return server, port


@pytest.fixture
def inventory_channel():
    server, port = _serve(
        lambda s: inventory_pb2_grpc.add_InventoryServiceServicer_to_server(InventoryService(), s))
    with grpc.insecure_channel(f"localhost:{port}") as ch:
        yield ch
    server.stop(grace=None)


@pytest.fixture
def orders_channel(inventory_channel):
    server, port = _serve(
        lambda s: orders_pb2_grpc.add_OrderServiceServicer_to_server(
            OrderService(inventory_channel), s))
    with grpc.insecure_channel(f"localhost:{port}") as ch:
        yield ch
    server.stop(grace=None)


@pytest.fixture
def token():
    return security.issue("test-user")

tests/test_system.py:

import grpc, pytest
from shop.v1 import common_pb2, inventory_pb2, inventory_pb2_grpc, orders_pb2, orders_pb2_grpc
from shared.interceptors import AuthInterceptor


def _items(*pairs):
    return [common_pb2.LineItem(sku=s, quantity=q) for s, q in pairs]


def test_stock_sufficient(inventory_channel):
    stub = inventory_pb2_grpc.InventoryServiceStub(inventory_channel)
    r = stub.CheckStock(inventory_pb2.CheckStockRequest(items=_items(("SKU-1", 2))))
    assert r.all_available is True
    assert r.results[0].available == 100


def test_stock_insufficient(inventory_channel):
    stub = inventory_pb2_grpc.InventoryServiceStub(inventory_channel)
    r = stub.CheckStock(inventory_pb2.CheckStockRequest(items=_items(("SKU-3", 1))))
    assert r.all_available is False


def test_empty_items_is_invalid_argument(inventory_channel):
    stub = inventory_pb2_grpc.InventoryServiceStub(inventory_channel)
    with pytest.raises(grpc.RpcError) as e:
        stub.CheckStock(inventory_pb2.CheckStockRequest(items=[]))
    assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT      # the contract


def test_client_streaming_restock(inventory_channel):
    stub = inventory_pb2_grpc.InventoryServiceStub(inventory_channel)
    def gen():
        yield inventory_pb2.RestockItem(sku="SKU-3", quantity=10)
        yield inventory_pb2.RestockItem(sku="SKU-3", quantity=5)
        yield inventory_pb2.RestockItem(sku="SKU-2", quantity=1)
    summary = stub.BulkRestock(gen())
    assert summary.items_processed == 3
    assert summary.skus_updated == 2


def test_create_order_confirmed(orders_channel):
    stub = orders_pb2_grpc.OrderServiceStub(orders_channel)
    r = stub.CreateOrder(orders_pb2.CreateOrderRequest(
        customer_id="C1", items=_items(("SKU-1", 2))))
    assert r.order.status == common_pb2.ORDER_STATUS_CONFIRMED
    assert r.order.total == pytest.approx(19.98)
    assert not r.order.HasField("rejection_reason")      # optional field: absent


def test_create_order_rejected_sets_reason(orders_channel):
    stub = orders_pb2_grpc.OrderServiceStub(orders_channel)
    r = stub.CreateOrder(orders_pb2.CreateOrderRequest(
        customer_id="C1", items=_items(("SKU-3", 1))))
    assert r.order.status == common_pb2.ORDER_STATUS_REJECTED
    assert r.order.HasField("rejection_reason")
    assert "SKU-3" in r.order.rejection_reason


def test_get_missing_order_is_not_found(orders_channel):
    stub = orders_pb2_grpc.OrderServiceStub(orders_channel)
    with pytest.raises(grpc.RpcError) as e:
        stub.GetOrder(orders_pb2.GetOrderRequest(id="nope"))
    assert e.value.code() == grpc.StatusCode.NOT_FOUND


def test_watch_orders_streams(orders_channel):
    stub = orders_pb2_grpc.OrderServiceStub(orders_channel)
    for _ in range(3):
        stub.CreateOrder(orders_pb2.CreateOrderRequest(
            customer_id="C9", items=_items(("SKU-1", 1))))
    got = list(stub.WatchOrders(orders_pb2.WatchOrdersRequest(customer_id="C9", limit=2)))
    assert len(got) == 2

tests/test_auth.py — the negative tests module 01 insisted on:

import grpc, pytest
from concurrent import futures
from shop.v1 import inventory_pb2, inventory_pb2_grpc, common_pb2
from services.inventory.service import InventoryService
from shared.interceptors import AuthInterceptor
from shared import security


@pytest.fixture
def secured():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=4),
                         interceptors=[AuthInterceptor(security.verify)])
    inventory_pb2_grpc.add_InventoryServiceServicer_to_server(InventoryService(), server)
    port = server.add_insecure_port("localhost:0")
    server.start()
    with grpc.insecure_channel(f"localhost:{port}") as ch:
        yield ch
    server.stop(grace=None)


def _req():
    return inventory_pb2.CheckStockRequest(
        items=[common_pb2.LineItem(sku="SKU-1", quantity=1)])


def test_no_token_rejected(secured):
    stub = inventory_pb2_grpc.InventoryServiceStub(secured)
    with pytest.raises(grpc.RpcError) as e:
        stub.CheckStock(_req())
    assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED


def test_garbage_token_rejected(secured):
    stub = inventory_pb2_grpc.InventoryServiceStub(secured)
    with pytest.raises(grpc.RpcError) as e:
        stub.CheckStock(_req(), metadata=[("authorization", "Bearer nonsense")])
    assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED


def test_valid_token_accepted(secured):
    stub = inventory_pb2_grpc.InventoryServiceStub(secured)
    md = [("authorization", f"Bearer {security.issue('u1')}")]
    assert stub.CheckStock(_req(), metadata=md).all_available is True

PYTHONPATH=. pytest -q

Step 11 — Benchmark it

TOKEN=$(PYTHONPATH=. python -c "from shared import security; print(security.issue('bench', 600))")

ghz --insecure --proto proto/shop/v1/inventory.proto --import-paths proto \
    --call shop.v1.InventoryService/CheckStock \
    -m "{\"authorization\":\"Bearer $TOKEN\"}" \
    -d '{"items":[{"sku":"SKU-1","quantity":1}]}' \
    -c 50 -n 20000 --connections 5 localhost:50052

hey -n 20000 -c 50 -m POST \
    -H "Authorization: Bearer $TOKEN" -H 'content-type: application/json' \
    -d '{"customer_id":"C1","items":[{"sku":"SKU-1","quantity":1}]}' \
    http://localhost:8000/v1/orders

Read the status distribution first (module 05). Then compare: the REST path traverses gateway → orders → inventory, so it is not a protocol comparison — it's an architecture comparison, and the extra hops should dominate. Note how much of the difference is hops rather than encoding.

Step 12 — Run it all together

docker-compose.yml:

services:
  inventory:
    build: .
    command: python services/inventory/main.py
    environment: { PYTHONPATH: /app, TLS: "1" }
    ports: ["50052:50052", "9101:9101"]
  orders:
    build: .
    command: python services/orders/main.py
    environment: { PYTHONPATH: /app, TLS: "1", INVENTORY_ADDR: "inventory:50052" }
    depends_on: [inventory]
    ports: ["50051:50051", "9102:9102"]
  gateway:
    build: .
    command: uvicorn services.gateway.main:app --host 0.0.0.0 --port 8000
    environment: { PYTHONPATH: /app, TLS: "1", ORDERS_ADDR: "orders:50051" }
    depends_on: [orders]
    ports: ["8000:8000"]

The certs from step 6 include DNS:inventory and DNS:orders precisely so mTLS validates against Docker's service names. Run TLS=1 and confirm the whole chain still works — that's the payoff for issuing SANs carefully.

Verify you've built it correctly

Work through this checklist; each item maps to a module:

• An unauthenticated CreateOrder returns UNAUTHENTICATED, and the handler never runs (01)
• grpc.health.v1.Health/Check works without a token (01, 02)
• Ordering out-of-stock SKU-3 yields REJECTED with rejection_reason set, and HasField is False when in stock (00)
• WatchOrders streams and the interceptor logs a non-zero duration and message count (03)
• localhost:9101/metrics shows grpc_server_handled_total split by code (03)
• curl on a missing order returns HTTP 404, not 200 (04)
• SIGTERM drains in-flight RPCs instead of severing them (02)
• pytest -q passes, including all three auth negative tests (03)
• TLS=1 works end to end under docker-compose (01)

Extend it yourself

No solutions given — each maps to a module you've finished:

1. Schema evolution under load. Add optional string coupon_code to CreateOrderRequest, deploy the new inventory service while the old orders service is still running, and prove nothing breaks (00).
2. Break it on purpose. Change LineItem.quantity from int32 to sint32, keep the field number, and observe what reaches the server (00).
3. Real load balancing. Run three inventory replicas on different ports, point orders at all three, and prove round_robin distributes while pick_first doesn't (02).
4. Retries. Add a gRPC retry policy via service config for UNAVAILABLE, and make sure CreateOrder is idempotent first — or you'll double-charge (02, and track 10).
5. Tracing. Add OpenTelemetry and get one trace spanning gateway → orders → inventory (03).
6. gRPC-Web. Put Envoy in front of orders and call it from a browser (04).
7. Deadline propagation. Make the gateway's 5 s budget shrink as it passes down, so inventory never outlives the client's patience (02, 03).

This is the end of the track

Modules 00-06 are complete. You've taken gRPC from "I can define a service" to schema evolution that survives production, mTLS, health and graceful shutdown, load balancing that actually balances, real observability, a REST gateway, honest benchmarks, and a working multi-service system that uses all of it.

Back to the track index: ../README.md Back to the backend master index: ../../README.md