Python Schema Registry Client

Warnings

• gotcha The `SchemaRegistryClient` may not be picklable, which can cause issues in distributed computing environments like Apache Spark or when using multiprocessing that relies on pickling objects. This is often due to its internal use of `requests.Session` (or `httpx.Client`).
  Fix: Avoid passing `SchemaRegistryClient` instances directly between processes. Instead, initialize a new client instance within each process or task where it's needed. For PySpark, consider using broadcast variables for configuration and re-initializing the client in UDFs.
• gotcha When using the `faust` extra (`pip install python-schema-registry-client[faust]`), the library pulls in `faust-streaming`, which is a fork of the original `faust` library. This might lead to version conflicts or unexpected behavior if your project already depends on a specific version of `faust`.
  Fix: If you require a specific `faust` version or wish to avoid the `faust-streaming` fork, install `faust` manually first, then install `python-schema-registry-client` without the `[faust]` extra: `pip install faust` followed by `pip install python-schema-registry-client`.
• gotcha SSL certificate verification can fail, especially when connecting to Schema Registry instances with self-signed certificates or improperly configured CA chains. This manifests as `requests.exceptions.SSLError` (or similar `httpx` errors).
  Fix: Ensure `ca_location`, `cert_location`, and `key_location` are correctly configured with paths to valid certificates. For development or controlled environments, `verify=False` can be passed to the underlying HTTP client (if exposed, or by subclassing/monkey-patching carefully) but is NOT recommended for production. Refer to `httpx` or `requests` SSL documentation for advanced configuration.
• gotcha Schema compatibility violations (e.g., error code 409) occur when registering a new schema that is incompatible with previous versions based on the subject's compatibility level.
  Fix: Always test compatibility before registering a new schema using `client.test_compatibility(subject_name, new_schema)`. Ensure your schema changes adhere to the configured compatibility rules (e.g., `BACKWARD`, `FORWARD`, `FULL`). Consider adding default values to new fields in Avro schemas to maintain backward compatibility.

Install

• pip install python-schema-registry-client Standard installation
• pip install python-schema-registry-client[faust] With Faust integration

Imports

• SchemaRegistryClient

  from schema_registry.client import SchemaRegistryClient

• AsyncSchemaRegistryClient

  from schema_registry.client import AsyncSchemaRegistryClient

• AvroSchema

  from schema_registry.client.schema import AvroSchema

• JsonSchema

  from schema_registry.client.schema import JsonSchema

• MessageSerializer

  from schema_registry.serializers import MessageSerializer

Quickstart

This quickstart demonstrates how to initialize the `SchemaRegistryClient`, define an Avro schema, register it with the Schema Registry, and then retrieve it by its ID and subject name. Ensure a Confluent Schema Registry instance is running and accessible at the specified `SCHEMA_REGISTRY_URL` (default: http://localhost:8081).

import os
import asyncio
from schema_registry.client import SchemaRegistryClient
from schema_registry.client.schema import AvroSchema

SCHEMA_REGISTRY_URL = os.environ.get('SCHEMA_REGISTRY_URL', 'http://localhost:8081')

async def main():
    client = SchemaRegistryClient(url=SCHEMA_REGISTRY_URL)

    avro_schema_definition = {
        "type": "record",
        "namespace": "com.example",
        "name": "SensorReading",
        "fields": [
            {"name": "id", "type": "string"},
            {"name": "value", "type": "int"}
        ]
    }
    
    # Create an AvroSchema object
    avro_schema = AvroSchema(avro_schema_definition)

    # Define a subject name
    subject = "sensor-readings-value"

    try:
        # Register the schema
        registered_schema = client.register(subject, avro_schema)
        print(f"Schema registered with ID: {registered_schema.schema_id}")

        # Get the schema by ID
        retrieved_schema = client.get_by_id(registered_schema.schema_id)
        print(f"Retrieved schema (ID {registered_schema.schema_id}): {retrieved_schema.schema.to_dict()}")

        # Get the latest schema for a subject
        latest_schema_info = client.get_latest_version(subject)
        print(f"Latest schema for '{subject}' (version {latest_schema_info.version}, ID {latest_schema_info.schema.schema_id}): {latest_schema_info.schema.to_dict()}")

    except Exception as e:
        print(f"An error occurred: {e}")

if __name__ == "__main__":
    # For synchronous use, just call client methods directly.
    # The example uses async to show client.register which is an async method by default
    # if the client is AsyncSchemaRegistryClient. For SchemaRegistryClient, it's synchronous.
    # However, the example above will work for both if using the sync client
    # due to how the `register` method is implemented (it's not truly async with await here).
    # Let's adjust for clarity to ensure it runs correctly with the default SchemaRegistryClient.
    # For a truly async example, one would use AsyncSchemaRegistryClient and 'await'.
    # For this quickstart, we'll keep it simple with the synchronous client.
    
    # Rerunning the quickstart for synchronous client clarification
    sync_client = SchemaRegistryClient(url=SCHEMA_REGISTRY_URL)
    
    avro_schema_definition = {
        "type": "record",
        "namespace": "com.example",
        "name": "SensorReading",
        "fields": [
            {"name": "id", "type": "string"},
            {"name": "value", "type": "int"}
        ]
    }
    
    avro_schema = AvroSchema(avro_schema_definition)
    subject = "sensor-readings-value-sync"

    try:
        registered_schema = sync_client.register(subject, avro_schema)
        print(f"Sync: Schema registered with ID: {registered_schema.schema_id}")

        retrieved_schema = sync_client.get_by_id(registered_schema.schema_id)
        print(f"Sync: Retrieved schema (ID {registered_schema.schema_id}): {retrieved_schema.schema.to_dict()}")

        latest_schema_info = sync_client.get_latest_version(subject)
        print(f"Sync: Latest schema for '{subject}' (version {latest_schema_info.version}, ID {latest_schema_info.schema.schema_id}): {latest_schema_info.schema.to_dict()}")

    except Exception as e:
        print(f"Sync: An error occurred: {e}")