avro-validator

raw JSON →
1.2.1 verified Fri May 01 auth: no python maintenance

A pure Python Avro schema validator that validates Python dicts against Avro schemas, supporting logical types, recursive types, custom types, and JSON Schema format for union types. Current version 1.2.1, released June 2022; maintenance mode.

pip install avro-validator
error AttributeError: module 'avro_validator' has no attribute 'validate'
cause Trying to call validate as a top-level function instead of method on AvroValidator instance.
fix
validator = AvroValidator(schema); result = validator.validate(data)
error avro_validator.exceptions.InvalidSchemaError: invalid literal for int() with base 10: ...
cause Schema field type is a custom type name that is not defined in the schema's 'types' list.
fix
Ensure all custom types are included in the schema's 'types' or use fully qualified names.
error TypeError: 'AvroResult' object is not iterable
cause Attempting to iterate over the AvroResult object (e.g., for error in result).
fix
Use result.errors to get the list of errors.
gotcha validate() returns an AvroResult object, not a boolean. Access .is_valid and .errors.
fix Use result.is_valid and result.errors instead of comparing result directly.
gotcha AvroSchema expects a Python dict, not a JSON string. If you have a JSON string, parse it with json.loads first.
fix schema_dict = json.loads(schema_str); schema = AvroSchema(schema_dict)
gotcha The library does not validate field defaults against the schema. A default with wrong type passes validation.
fix Manually check defaults before validation or use a different library if strict default validation is needed.

Create an Avro schema, validate a record, and check result.

from avro_validator import AvroValidator, AvroSchema

schema = AvroSchema({
    'type': 'record',
    'name': 'User',
    'fields': [{'name': 'name', 'type': 'string'}, {'name': 'age', 'type': 'int'}]
})
validator = AvroValidator(schema)

data = {'name': 'Alice', 'age': 30}
result = validator.validate(data)
print(result.is_valid)  # True
print(result.errors)    # []