Register an Avro schema against the Confluent Schema Registry

register_schema-tls.py

#!/usr/bin/python

import os
import sys

import requests

schema_registry_url = sys.argv[1]
topic = sys.argv[2]
schema_file = sys.argv[3]
cert_file_path = sys.argv[4] #your certificate format pem
key_file_path = sys.argv[5] #your key format .key

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)
print("Cert file: " + cert_file_path)
print("Key file: " + key_file_path)
print

with open(aboslute_path_to_schema, 'r') as content_file:
    schema = content_file.read()

payload = "{\"schema\":\""\
          + schema.replace("\"", "\\\"").replace("\t", "").replace("\n", "").replace(" ", "") \
          + "\"}"

print("payload: " + payload)

url = schema_registry_url + "/subjects/" + topic + "-value/versions"
headers = {"Content-Type": "application/vnd.schemaregistry.v1+json"}

cert = (cert_file_path, key_file_path)
r = requests.post(url, headers=headers, cert=cert, data=payload.encode('utf-8'), verify=False)

print("request: " , r.status_code)

if r.status_code == requests.codes.ok:
    print("Success")
else:
    r.raise_for_status()