confluent_schema_registry_client_avro_schema_register_retrieve.py

1from schema_registry.client import SchemaRegistryClient, schema
2
3# Initialize the client
4client = SchemaRegistryClient(url="http://localhost:8081")
5
6# Define an Avro schema
7avro_user_schema = schema.AvroSchema({
8    "type": "record",
9    "name": "User",
10    "fields": [
11        {"name": "name", "type": "string"},
12        {"name": "age", "type": "int"}
13    ]
14})
15
16# Register the schema under a subject
17subject = "user-value"
18schema_id = client.register(subject, avro_user_schema)
19
20print(f"Schema registered with ID: {schema_id}")
21
22# Get schema by ID
23registered_schema = client.get_by_id(schema_id)
24print(f"Retrieved schema: {registered_schema.schema_str}")
25
26# Get the latest version of a schema for a subject
27schema_version = client.get_schema(subject)
28print(f"Latest version: {schema_version.version}")