SchemaRegistryClient Class

SchemaRegistryClient is a client for registering and retrieving schemas from the Azure Schema Registry service.

Inheritance

builtins.object

SchemaRegistryClient

Constructor

SchemaRegistryClient(fully_qualified_namespace: str, credential: AsyncTokenCredential, **kwargs: Any)

Parameters

fully_qualified_namespace

str

Required

The Schema Registry service fully qualified host name. For example: my-namespace.servicebus.windows.net.

credential

AsyncTokenCredential

Required

To authenticate managing the entities of the SchemaRegistry namespace.

api_version

str

The Schema Registry service API version to use for requests. Default value and only accepted value currently is "2021-10".

Examples

Create a new instance of the SchemaRegistryClient.

   SCHEMAREGISTRY_FQN = os.environ["SCHEMAREGISTRY_FULLY_QUALIFIED_NAMESPACE"]
   token_credential = DefaultAzureCredential()
   schema_registry_client = SchemaRegistryClient(
       fully_qualified_namespace=SCHEMAREGISTRY_FQN, credential=token_credential
   )

Methods

close	This method is to close the sockets opened by the client. It need not be used when using with a context manager.
get_schema	Gets a registered schema by its unique ID. Azure Schema Registry guarantees that ID is unique within a namespace.
get_schema_properties	Gets the schema properties corresponding to an existing schema within the specified schema group, as matched by schema defintion comparison.
register_schema	Register new schema. If schema of specified name does not exist in specified group, schema is created at version 1. If schema of specified name exists already in specified group, schema is created at latest version + 1.

close

This method is to close the sockets opened by the client. It need not be used when using with a context manager.

async close() -> None

get_schema

Gets a registered schema by its unique ID. Azure Schema Registry guarantees that ID is unique within a namespace.

async get_schema(schema_id: str, **kwargs: Any) -> azure.schemaregistry._common._schema.Schema

Parameters

schema_id

str

Required

References specific schema in registry namespace.

Return type

Schema

Exceptions

HttpResponseError

Examples

Get schema by id.

   schema = await schema_registry_client.get_schema(schema_id)
   definition = schema.definition
   properties = schema.properties

get_schema_properties

Gets the schema properties corresponding to an existing schema within the specified schema group, as matched by schema defintion comparison.

async get_schema_properties(group_name: str, name: str, definition: str, format: Union[str, azure.schemaregistry._common._constants.SchemaFormat], **kwargs: Any) -> azure.schemaregistry._common._schema.SchemaProperties

Parameters

group_name

str

Required

Schema group under which schema should be registered.

name

str

Required

Name of schema for which properties should be retrieved.

definition

str

Required

String representation of the schema for which properties should be retrieved.

format

<xref:Union>[str, SchemaFormat]

Required

Format for the schema for which properties should be retrieved.

Return type

SchemaProperties

Exceptions

HttpResponseError

Examples

Get schema by id.

   group_name = os.environ["SCHEMAREGISTRY_GROUP"]
   name = "your-schema-name"
   format = "Avro"
   schema_json = {
       "namespace": "example.avro",
       "type": "record",
       "name": "User",
       "fields": [
           {"name": "name", "type": "string"},
           {"name": "favorite_number", "type": ["int", "null"]},
           {"name": "favorite_color", "type": ["string", "null"]},
       ],
   }
   definition = json.dumps(schema_json, separators=(",", ":"))
   schema_properties = await schema_registry_client.get_schema_properties(
       group_name, name, definition, format
   )
   schema_id = schema_properties.id

register_schema

Register new schema. If schema of specified name does not exist in specified group, schema is created at version 1. If schema of specified name exists already in specified group, schema is created at latest version + 1.

async register_schema(group_name: str, name: str, definition: str, format: Union[str, azure.schemaregistry._common._constants.SchemaFormat], **kwargs: Any) -> azure.schemaregistry._common._schema.SchemaProperties

Parameters

group_name

str

Required

Schema group under which schema should be registered.

name

str

Required

Name of schema being registered.

definition

str

Required

String representation of the schema being registered.

format

<xref:Union>[str, SchemaFormat]

Required

Format for the schema being registered. For now Avro is the only supported schema format by the service.

Return type

SchemaProperties

Exceptions

HttpResponseError

Examples

Register a new schema.

   GROUP_NAME = os.environ["SCHEMAREGISTRY_GROUP"]
   NAME = "your-schema-name"
   FORMAT = "Avro"
   SCHEMA_JSON = {
       "namespace": "example.avro",
       "type": "record",
       "name": "User",
       "fields": [
           {"name": "name", "type": "string"},
           {"name": "favorite_number", "type": ["int", "null"]},
           {"name": "favorite_color", "type": ["string", "null"]},
       ],
   }
   DEFINITION = json.dumps(SCHEMA_JSON, separators=(",", ":"))
   schema_properties = await schema_registry_client.register_schema(
       GROUP_NAME, NAME, DEFINITION, FORMAT
   )
   schema_id = schema_properties.id