Azure Schema Registry Client Library

1.3.0 · active · verified Sat Apr 11

The Azure Schema Registry client library for Python (version 1.3.0) provides a robust interface for interacting with the Azure Schema Registry service. This service, hosted by Azure Event Hubs, offers a centralized repository for managing, versioning, and storing schemas. The library enables applications to register and retrieve schemas, and includes a JSON schema-based encoder for efficient payload encoding and decoding using schema identifiers rather than full schema definitions. It adheres to the Azure SDK guidelines, offering a consistent and Pythonic experience. It is actively maintained with a regular release cadence as part of the broader Azure SDK for Python.

Warnings

breaking Python 3.7 is no longer supported with `azure-schemaregistry` version 1.3.0 and later. Python 3.8 or higher is now required.
Fix: Upgrade your Python environment to 3.8 or later.
breaking Several API methods and parameters were renamed or reordered in earlier stable releases. Specifically, `get_schema_id` was renamed to `get_schema_properties`, the `schema_id` parameter in `get_schema` was renamed to `id`, and the `register_schema` and `get_schema_properties` methods had parameter reordering and renaming (`schema_group` to `group_name`, `schema_name` to `name`).
Fix: Update method and parameter names according to the current API reference. Refer to the Changelog for specific version details.
gotcha The `fully_qualified_namespace` for the `SchemaRegistryClient` must be in the format `<yournamespace>.servicebus.windows.net/`. Incorrect formats (e.g., including `sb://`) will lead to connection errors or 'Bad Request' responses.
Fix: Ensure the namespace string strictly follows the format: `your-eventhub-namespace.servicebus.windows.net`.
gotcha Authentication requires the `azure-identity` package and proper Azure Active Directory (AAD) setup. If `DefaultAzureCredential` fails, verify your environment variables (`AZURE_TENANT_ID`, `AZURE_CLIENT_ID`, `AZURE_CLIENT_SECRET`) or Managed Identity configuration. Role-Based Access Control (RBAC) permissions on the Schema Registry are crucial.
Fix: Install `azure-identity` (`pip install azure-identity`), configure appropriate environment variables or managed identity, and ensure your AAD principal has necessary permissions (e.g., 'Schema Registry Contributor' or 'Schema Registry Reader').
gotcha When using `JsonSchemaEncoder`, encoding/decoding errors (e.g., invalid content or content type) will raise `azure.schemaregistry.encoder.jsonencoder.InvalidContentError`. The `content_type` must adhere to the format `application/json;serialization=Json+<schema ID>`.
Fix: Validate your input content against the registered schema and ensure the `content_type` header is correctly formatted, especially the embedded schema ID.

Install

pip install azure-schemaregistry Base installation
pip install azure-schemaregistry[jsonencoder] With JSON schema validation support

Imports

SchemaRegistryClient

from azure.schemaregistry import SchemaRegistryClient

JsonSchemaEncoder
```
from azure.schemaregistry.encoder.jsonencoder import JsonSchemaEncoder
```
The JsonSchemaEncoder is nested under 'encoder.jsonencoder' for version 1.3.0 and later.
DefaultAzureCredential
```
from azure.identity import DefaultAzureCredential
```
Required for authenticating with Azure services. Part of the 'azure-identity' package.

Quickstart

This quickstart demonstrates how to instantiate a `SchemaRegistryClient` using `DefaultAzureCredential` and then register and retrieve a JSON schema. Ensure that `SCHEMAREGISTRY_FULLY_QUALIFIED_NAMESPACE`, `SCHEMAREGISTRY_GROUP_NAME`, and `SCHEMAREGISTRY_SCHEMA_NAME` environment variables are set, along with Azure AD credentials (`AZURE_TENANT_ID`, `AZURE_CLIENT_ID`, `AZURE_CLIENT_SECRET`) for `DefaultAzureCredential` to function.

import os
from azure.schemaregistry import SchemaRegistryClient
from azure.identity import DefaultAzureCredential

# Set environment variables for authentication and Schema Registry endpoint
# AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET for DefaultAzureCredential
# SCHEMAREGISTRY_FULLY_QUALIFIED_NAMESPACE for the Schema Registry endpoint

fully_qualified_namespace = os.environ.get('SCHEMAREGISTRY_FULLY_QUALIFIED_NAMESPACE', 'your-namespace.servicebus.windows.net')
schema_group = os.environ.get('SCHEMAREGISTRY_GROUP_NAME', 'my-schema-group')
schema_name = os.environ.get('SCHEMAREGISTRY_SCHEMA_NAME', 'my-json-schema')
schema_definition = '''{
    "type": "object",
    "$id": "https://example.com/example.json",
    "title": "Person",
    "description": "A person schema",
    "properties": {
        "name": {
            "type": "string",
            "description": "The person's name."
        },
        "age": {
            "type": "integer",
            "description": "The person's age.",
            "minimum": 0
        }
    },
    "required": ["name"]
}'''
schema_format = "Json"

credential = DefaultAzureCredential()
client = SchemaRegistryClient(fully_qualified_namespace, credential)

try:
    # Register a schema
    schema_properties = client.register_schema(
        group_name=schema_group,
        name=schema_name,
        format=schema_format,
        definition=schema_definition
    )
    print(f"Registered schema with ID: {schema_properties.id}")
    print(f"Schema Version: {schema_properties.version}")

    # Get a schema by its ID
    retrieved_schema = client.get_schema(schema_properties.id)
    print(f"Retrieved schema name: {retrieved_schema.name}")

except Exception as e:
    print(f"An error occurred: {e}")
finally:
    # Always close the credential when done (if using certain types like ManagedIdentityCredential)
    # DefaultAzureCredential handles this contextually.
    pass

view raw JSON →