A Python client and resolver for [Iglu schema repositories][iglu-wiki]
pip install iglu_client
importjsonimportjsonschemafromiglu_clientimportSchemaResolver, SelfDescribingJsonresolver_conf="""{ "schema": "iglu:com.snowplowanalytics.iglu/resolver-config/jsonschema/1-0-2", "data": { "cacheSize": 500, "repositories": [ { "name": "My Private Repo", "priority": 0, "vendorPrefixes": ["com.mycompany"], "connection": {"http": {"uri": "http://myprivaterepo.com", "apikey": "api-key"}} }, { "name": "Iglu Central", "priority": 1, "vendorPrefixes": ["com.snowplowanalytics.snowplow"], "connection": {"http": {"uri": "http://iglucentral.com"}} } ] }}"""resolver=SchemaResolver.parse(resolver_conf)
# Retrieve a schemaschema=resolver.lookup_schema("iglu:com.snowplowanalytics.snowplow/link_click/jsonschema/1-0-0")
print("schema: %s"%schema)
# Validate a self-describing JSONevent="""{ "schema": "iglu:com.snowplowanalytics.snowplow/link_click/jsonschema/1-0-0", "data": { "targetUrl": "https://mixpanel.com" }}"""data=SelfDescribingJson.parse(event)
try:
data.validate(resolver) # will throw an exception if the event is invalidprint("event is valid")
exceptjsonschema.exceptions.ValidationErrorase:
print("event is invalid")
ifdata.valid(resolver): # will return True or Falseprint("event is valid")
else:
print("event is invalid")