As of now i am doing something like this reading avsc file to get schema
value_schema = avro.load('client.avsc')
can i do something to get schema from confluent schema registry using topic-name?
i found one way but didn't figure out how to use it.
https://github.com/marcosschroh/python-schema-registry-client
Using confluent-kafka-python
from confluent_kafka.avro.cached_schema_registry_client import CachedSchemaRegistryClient
sr = CachedSchemaRegistryClient({
'url': 'http://localhost:8081',
'ssl.certificate.location': '/path/to/cert', # optional
'ssl.key.location': '/path/to/key' # optional
})
value_schema = sr.get_latest_schema("orders-value")[1]
key_schema= sr.get_latest_schema("orders-key")[1]
Using SchemaRegistryClient
Getting schema by subject name
from schema_registry.client import SchemaRegistryClient
sr = SchemaRegistryClient('localhost:8081')
my_schema = sr.get_schema(subject='mySubject', version='latest')
Getting schema by ID
from schema_registry.client import SchemaRegistryClient
sr = SchemaRegistryClient('localhost:8081')
my_schema = sr.get_by_id(schema_id=1)
you can use get_latest_version function to get schema information
from confluent_kafka.schema_registry import SchemaRegistryClient
sr = SchemaRegistryClient({"url": 'http://localhost:8081'})
subjects = sr.get_subjects()
for subject in subjects:
schema = sr.get_latest_version(subject)
print(schema.version)
print(schema.schema_id)
print(schema.schema.schema_str)I did like this it worked for me
import requests
import os
SCHEMA_REGISTRY_URL = os.getenv('SCHEMA_REGISTRY_URL');
print("SCHEMA_REGISTRY_URL: ", SCHEMA_REGISTRY_URL)
URL = SCHEMA_REGISTRY_URL + '/subjects/' + topic + '/versions/latest/schema'
r = requests.get(url=URL)
schema = r.json()
print("Schema From Schema Registry ==========================>>")
print("Schema: ", schema)
