How to Mock a Schema Registry Avro Record in Python

Encode a Python dict into Avro binary using an inline schema, mimicking a schema registry record for tests or mocks.

Easy Python 3.9+ Aug 9, 2026 Microservices patterns 15 views 0 copies

avro schema-registry serialization kafka mock

Requires third-party packages — install first

pip install avro

Python code

26 lines

Python 3.9+

import io
from avro.schema import parse
from avro.io import DatumWriter, BinaryEncoder

schema_json = """
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "int"},
    {"name": "email", "type": ["null", "string"], "default": null}
  ]
}
"""

schema = parse(schema_json)
record = {"name": "Alice", "age": 30}

writer = DatumWriter(schema)
bytes_writer = io.BytesIO()
encoder = BinaryEncoder(bytes_writer)
writer.write(record, encoder)

print(f"Encoded {len(bytes_writer.getvalue())} bytes")
print(f"Bytes: {bytes_writer.getvalue()}")

Output

stdout

Encoded 11 bytes
Bytes: b'\x06Alice\x3c\x00'

How it works

The avro.schema.parse converts a JSON schema string into an Avro schema object. DatumWriter writes a Python dict to binary using that schema. BinaryEncoder wraps a BytesIO stream to capture the encoded output. The record omits the optional email field, so the encoder writes the default null. The byte count reflects the string length prefix, the string bytes, and the int zigzag encoding.

Common mistakes

Forgetting the default for a nullable field when your dict omits it.
Using `json.dumps` instead of the Avro BinaryEncoder for the payload.
Passing a raw dict to `DatumWriter` without a parsed schema.

Variations

Use `avro.io.DatumReader` to decode the bytes back into a dict.
Load a schema from a file or schema registry client instead of an inline string.

Real-world use cases

Unit-testing Kafka producers by encoding a mocked Avro record without contacting a schema registry.
Generating binary Avro fixtures for integration tests of consumers expecting schema-registry wire format.
Simulating service-to-service payloads when developing a new microservice offline.

How to Mock a Schema Registry Avro Record in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Microservices patterns

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Microservices patterns

Keep learning

Tutorials

Quizzes