Azure Schema Registry Avro Encoder Client Library for Python

1.0.0 verified Fri May 01 auth: no python

Microsoft Azure Schema Registry Avro Encoder client library for Python (v1.0.0) enables encoding/decoding Avro data with schema stored in Azure Schema Registry. It is built on Apache Avro (<1.12) and requires Python >=3.7. Release cadence is monthly.

pip install azure-schemaregistry-avroencoder

Common errors

error ImportError: cannot import name 'AvroEncoder' from 'azure.schemaregistry.avroencoder' ↓

cause Incorrect import path missing 'encoder' subpackage.

fix

Use: from azure.schemaregistry.encoder.avroencoder import AvroEncoder

error AvroException: 'int' object has no attribute 'encode' ↓

cause Passing a single value instead of a dict with schema fields.

fix

Ensure data is a dictionary matching the Avro schema record.

Warnings

breaking Avro library version: This library requires avro (Apache Avro) version >=1.10.0, <1.12.0. Using avro>=1.12 will cause import errors. ↓

fix Pin avro to >=1.10.0,<1.12.0 in your requirements.

gotcha AvroEncoder is not thread-safe for encode/decode operations due to internal schema cache. Create a new instance per thread or use locks. ↓

fix Create a new AvroEncoder per thread or synchronize access.

deprecated Python 3.6 support ended. The library requires Python >=3.7 for this version. ↓

fix Upgrade Python to >=3.7.

Imports

AvroEncoder

wrong

from azure.schemaregistry.avroencoder import AvroEncoder

correct

from azure.schemaregistry.encoder.avroencoder import AvroEncoder

Correct import includes 'encoder' subpackage

Quickstart

Encode and decode Avro data with schema registry.

import os
from azure.schemaregistry import SchemaRegistryClient
from azure.schemaregistry.encoder.avroencoder import AvroEncoder
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
schema_registry_client = SchemaRegistryClient(
    fully_qualified_namespace=os.environ.get('SCHEMA_REGISTRY_FQDN', 'example.servicebus.windows.net'),
    credential=credential
)
encoder = AvroEncoder(client=schema_registry_client, group_name=os.environ.get('GROUP_NAME', 'my-group'))

# Encode
data = {"name": "Alice", "age": 30}
schema = {
    "type": "record",
    "name": "User",
    "fields": [
        {"name": "name", "type": "string"},
        {"name": "age", "type": "int"}
    ]
}
encoded = encoder.encode(data, schema=schema)
print(f"Encoded bytes: {encoded}")

# Decode
decoded = encoder.decode(encoded)
print(f"Decoded data: {decoded}")