Type‑Safe Polyglot Microservices: Sharing Schemas Between TypeScript and Python with Protobuf, Zod, and Runtime Validation

Introduction

Microservice architectures often span several programming languages. A common source of bugs is the schema drift that appears when the contract that a service publishes (e.g., a JSON payload) diverges from what its consumers expect.

In a JavaScript‑heavy stack you can lean on TypeScript’s type system, while in a data‑science‑oriented service you might be writing pure Python. This article shows a pragmatic, production‑ready approach that gives you:

• Single source of truth – one .proto file per domain object.
• Compile‑time safety in the TypeScript code base via protoc‑gen‑ts and Zod schemas that mirror the generated types.
• Runtime validation in Python using protobuf’s generated classes plus a tiny pydantic‑style validator that guarantees the same constraints enforced by Zod.

The result is a polyglot contract that is type‑safe on both ends without duplicating validation logic.

TL;DR – Define a protobuf schema, generate TypeScript and Python code, add Zod wrappers for TS, add a small runtime validator for Python, and you get end‑to‑end guarantees that your services speak the same language.

1. Why Protobuf + Zod + Runtime Validation?

Protobuf solves the transport problem, Zod solves the client‑side validation problem, and a tiny Python validator mirrors the same rules on the server side.

2. Defining the Contract

Create a directory schemas/ that lives at the root of the monorepo.

// schemas/order.proto
syntax = "proto3";

package orders;

// A monetary value expressed in cents to avoid floating point issues.
message Money {
  int64 amount = 1;   // amount in cents, must be >= 0
  string currency = 2; // ISO‑4217 code, e.g. "USD"
}

// An item inside an order.
message OrderItem {
  string sku = 1;
  int32 quantity = 2; // > 0
  Money unit_price = 3;
}

// The top‑level order message.
message Order {
  string id = 1;
  repeated OrderItem items = 2;
  Money total = 3;
  string customer_email = 4;
  // Timestamp in seconds since epoch.
  int64 placed_at = 5;
}

Only structural constraints (required/optional, repeated) belong in protobuf. Business rules like “quantity must be positive” are expressed later with Zod and Python validators.

Run the compiler once:

# TypeScript generation
protoc --plugin=protoc-gen-ts=$(npm bin)/protoc-gen-ts \
  --js_out=import_style=commonjs,binary:./ts/src/generated \
  --ts_out=grpc_js:./ts/src/generated \
  -I schemas schemas/order.proto

# Python generation
python -m grpc_tools.protoc -I schemas \
  --python_out=./py/src/generated \
  --grpc_python_out=./py/src/generated \
  schemas/order.proto

Now you have:

• ts/src/generated/order_pb.ts – raw protobuf classes.
• py/src/generated/order_pb2.py – raw protobuf classes.

3. Adding Zod Schemas (TypeScript)

Create a thin Zod wrapper that re‑uses the generated TypeScript types for static inference.

// ts/src/schemas/order.zod.ts
import { z } from "zod";
import type {
  Money as MoneyProto,
  OrderItem as OrderItemProto,
  Order as OrderProto,
} from "./generated/order_pb";

// Helper to keep Zod ↔ protobuf type mapping tidy
type Infer<T extends z.ZodTypeAny> = z.infer<T>;

export const MoneySchema = z.object({
  amount: z.bigint().nonnegative(), // protobuf int64 → bigint in TS
  currency: z.string().length(3),   // ISO‑4217
});

export const OrderItemSchema = z.object({
  sku: z.string().min(1),
  quantity: z.number().int().positive(),
  unit_price: MoneySchema,
});

export const OrderSchema = z.object({
  id: z.string().uuid(),
  items: z.array(OrderItemSchema).min(1),
  total: MoneySchema,
  customer_email: z.string().email(),
  placed_at: z.bigint().refine((v) => v > 0n, { message: "must be a valid epoch" }),
});

/**
 * Helper to convert a protobuf message to a plain object that Zod can validate.
 * Protobuf generated classes expose getters like `getId()`.
 */
export function toPlainOrder(pb: OrderProto): Infer<typeof OrderSchema> {
  const items = pb.getItemsList().map((it) => ({
    sku: it.getSku(),
    quantity: it.getQuantity(),
    unit_price: {
      amount: BigInt(it.getUnitPrice()!.getAmount()),
      currency: it.getUnitPrice()!.getCurrency(),
    },
  }));

  return {
    id: pb.getId(),
    items,
    total: {
      amount: BigInt(pb.getTotal()!.getAmount()),
      currency: pb.getTotal()!.getCurrency(),
    },
    customer_email: pb.getCustomerEmail(),
    placed_at: BigInt(pb.getPlacedAt()),
  };
}

/**
 * Runtime guard used in request handlers, background jobs, etc.
 */
export function assertValidOrder(pb: OrderProto): asserts pb is OrderProto {
  OrderSchema.parse(toPlainOrder(pb));
}

Key points

• z.bigint() mirrors protobuf int64.
• assertValidOrder throws on the first violation, giving a clear error stack.
• The schema is source‑of‑truth for the TS side; any change in the business rule requires editing only the Zod definition.

4. Python Runtime Validation

Python’s protobuf objects are essentially data containers, but they lack built‑in validation beyond type coercion. We add a small validator that mirrors the Zod rules.

# py/src/validation/order.py
from __future__ import annotations
from dataclasses import dataclass
from typing import List
import re
from google.protobuf.message import Message
from .generated import order_pb2

# Simple exception hierarchy
class ValidationError(ValueError):
    pass

def _validate_money(money: order_pb2.Money) -> None:
    if money.amount < 0:
        raise ValidationError("Money.amount must be non‑negative")
    if not re.fullmatch(r"[A-Z]{3}", money.currency):
        raise ValidationError("Money.currency must be a 3‑letter ISO‑4217 code")

def _validate_order_item(item: order_pb2.OrderItem) -> None:
    if not item.sku:
        raise ValidationError("OrderItem.sku cannot be empty")
    if item.quantity <= 0:
        raise ValidationError("OrderItem.quantity must be > 0")
    _validate_money(item.unit_price)

def validate_order(order: order_pb2.Order) -> None:
    """Raises ValidationError if the order does not satisfy business rules."""
    if not re.fullmatch(
        r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[4][0-9a-fA-F]{3}"
        r"-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}",
        order.id,
    ):
        raise ValidationError("Order.id must be a UUID")
    if len(order.items) == 0:
        raise ValidationError("Order must contain at least one item")
    for it in order.items:
        _validate_order_item(it)
    _validate_money(order.total)
    if not re.fullmatch(r"[^@]+@[^@]+\.[^@]+", order.customer_email):
        raise ValidationError("Invalid customer_email")
    if order.placed_at <= 0:
        raise ValidationError("placed_at must be a positive epoch timestamp")

The validator is deliberately thin – only the business rules we expressed in Zod. If you later add a rule (e.g., “currency must be USD or EUR”), edit both order.zod.ts and order.py in lock‑step. To avoid forgetting, we’ll add a CI lint step later.

5. Using the Contract in a Real Service

5.1 TypeScript – an HTTP endpoint (Express)

// ts/src/server/orders.ts
import express from "express";
import { Order } from "./generated/order_pb";
import { assertValidOrder, toPlainOrder } from "./schemas/order.zod";

const router = express.Router();

router.post("/orders", async (req, res) => {
  // Assume body is binary protobuf payload
  const buf = Buffer.from(req.body);
  const order = Order.deserializeBinary(buf);

  // 1️⃣ Runtime validation
  try {
    assertValidOrder(order);
  } catch (e) {
    return res.status(400).json({ error: (e as Error).message });
  }

  // 2️⃣ Business logic – e.g., store in DB
  const plain = toPlainOrder(order);
  await db.orders.insert(plain); // db layer expects plain JS object

  return res.status(201).json({ id: plain.id });
});

export default router;

5.2 Python – a background worker (Celery)

# py/src/workers/process_order.py
from celery import Celery
from .generated import order_pb2
from .validation.order import validate_order

app = Celery('orders', broker='redis://localhost:6379/0')

@app.task
def handle_order(payload: bytes):
    """Celery task that receives a protobuf-encoded Order."""
    order = order_pb2.Order()
    order.ParseFromString(payload)

    # Runtime validation – will raise ValidationError on bad data
    validate_order(order)

    # Convert to dict for ORM (SQLAlchemy example)
    data = {
        "id": order.id,
        "customer_email": order.customer_email,
        "placed_at": order.placed_at,
        # flatten items, etc.
    }
    # ... persist with SQLAlchemy ...

Both sides deserialize the same binary payload, run identical business constraints, and then continue with their own data‑access layers.

6. Keeping the Two Validators Synchronized

Manual copy‑pasting of rules is error‑prone. A lightweight solution is to generate the Python validator from the Zod schema using a custom script.

// scripts/generate_py_validator.ts
import { writeFileSync } from "fs";
import { OrderSchema } from "../ts/src/schemas/order.zod";
import { ZodObject, ZodString, ZodNumber, ZodBigInt } from "zod";

function emitPython(schema: ZodObject<any>, name: string): string {
  const lines: string[] = [];
  lines.push(`def validate_${name.lower()}(msg):`);
  lines.push(`    """Auto‑generated from Zod. Raises ValidationError."""`);
  // Simplified – in real script you’d handle all Zod types
  for (const [key, def] of Object.entries(schema.shape)) {
    if (def instanceof ZodString) {
      lines.push(`    if not isinstance(msg.${key}, str):`);
      lines.push(`        raise ValidationError("${key} must be a string")`);
    } else if (def instanceof ZodNumber) {
      lines.push(`    if not isinstance(msg.${key}, int):`);
      lines.push(`        raise ValidationError("${key} must be an int")`);
    } else if (def instanceof ZodBigInt) {
      lines.push(`    if not isinstance(msg.${key}, int):`);
      lines.push(`        raise ValidationError("${key} must be a non‑negative int")`);
    }
    // Add more branches for .positive(), .email(), etc.
  }
  lines.push(`    # Add further nested validation here`);
  return lines.join("\n");
}

// Generate file
const code = `
from .validation import ValidationError
${emitPython(OrderSchema, "Order")}
`;
writeFileSync("../py/src/validation/generated_order.py", code);

Run this script as part of the CI pipeline (npm run generate:validators). The generated Python file lives next to the hand‑written validators; you can import and reuse it. This approach offers single‑source‑of‑truth without heavy tooling.

7. Versioning Strategies

When a new field is added:

1. Add it to the .proto file with a new field number (never reuse numbers).
2. Regenerate both TS and Python code (protoc).
3. Extend the Zod schema and the Python validator.
4. Bump the package version (e.g., @myorg/orders-schema) so services can detect incompatibility early.

Because protobuf fields are optional by default, older services will ignore unknown fields, preserving backward compatibility.

8. Performance Considerations

• Binary payload – protobuf messages are typically 5‑10× smaller than JSON, reducing bandwidth.
• Validation cost – Zod parsing a plain object is cheap (< 0.2 ms for a 10‑item order). The Python validator runs in pure Python; for high‑throughput services you can compile it with Cython or use pydantic (which is already C‑optimized).
• Caching – If the same schema is validated many times per second, cache the compiled Zod schema (OrderSchema) and reuse the Python validator function (they are stateless).

9. Tooling Checklist

10. Takeaways

• One contract, two worlds – protobuf defines the wire format; Zod and a tiny Python validator enforce the same business rules.
• Zero duplication – generate code, share type information, and optionally generate the Python validator from Zod.
• Strong guarantees – compile‑time inference in TS, runtime errors in both languages, and protobuf’s built‑in compatibility rules keep services aligned as they evolve.

Adopting this pattern means you can confidently add a new microservice written in Python to a TypeScript‑first ecosystem (or vice‑versa) without fearing silent contract mismatches. The effort is a one‑time setup plus a disciplined CI step, and the payoff is a dramatically reduced bug surface in polyglot environments.