Skip to content

Instantly share code, notes, and snippets.

Last active October 18, 2022 08:26
What would you like to do?
Register an Avro schema against the Confluent Schema Registry
import os
import sys
import requests
schema_registry_url = sys.argv[1]
topic = sys.argv[2]
schema_file = sys.argv[3]
aboslute_path_to_schema = os.path.join(os.getcwd(), schema_file)
print("Schema Registry URL: " + schema_registry_url)
print("Topic: " + topic)
print("Schema file: " + schema_file)
with open(aboslute_path_to_schema, 'r') as content_file:
schema =
payload = "{ \"schema\": \"" \
+ schema.replace("\"", "\\\"").replace("\t", "").replace("\n", "") \
+ "\" }"
url = schema_registry_url + "/subjects/" + topic + "-value/versions"
headers = {"Content-Type": "application/vnd.schemaregistry.v1+json"}
r =, headers=headers, data=payload)
if r.status_code ==
Copy link

Please use in the first line


to avoid errors

Copy link

Please add a license that permits reuse. We want to use what you have and have an extension for security but can't without a license like

# Licensed under the Apache 2.0 license

as the second line.


Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment