Skip to content

Instantly share code, notes, and snippets.

@talevy
Created June 3, 2013 05:21
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save talevy/5696236 to your computer and use it in GitHub Desktop.
This is an example avro codegen'd python object representing a specific avro schema
#
# Autogenerated by Avro
#
# DO NOT EDIT DIRECTLY
#
from avro.io import validate
from avro.schema import parse
class Event(dict):
_schema = parse('''{"type":"record","name":"Event","namespace":"org.logging.api","fields":[{"name":"event_type","type":"string"},{"name":"random_string","type":["string","null"],"default":null},{"name":"event_id","type":["long","null"],"default":null},{"name":"user_id","type":["long","null"],"default":null}]}''')
@classmethod
def typecheck(cls, attr, value):
'''
Given an attribute and value, returns True iff this class's schema
includes that attribute and the value is of the attribute's specified type.
'''
field = cls._schema.fields_dict.get(attr)
return field and validate(field.type, value)
@classmethod
def is_valid(cls, attributes):
return all(cls.typecheck(k, v) for k, v in attributes.items())
@property
def event_type(self):
return self['event_type']
@event_type.setter
def event_type(self, value):
if Event.typecheck('event_type', value):
self['event_type'] = value
else:
raise TypeError("cannot set event_type to " + repr(type(value)) + ", expecting: str")
@property
def random_string(self):
return self['random_string']
@random_string.setter
def random_string(self, value):
if Event.typecheck('random_string', value):
self['random-string'] = value
else:
raise TypeError("cannot set random_string to " + repr(type(value)) + ", expecting: str")
@property
def event_id(self):
return self['event_id']
@event_id.setter
def event_id(self, value):
if Event.typecheck('event_id', value):
self['event_id'] = value
else:
raise TypeError("cannot set event_id to " + repr(type(value)) + ", expecting: long")
@property
def user_id(self):
return self['user_id']
@user_id.setter
def user_id(self, value):
if Event.typecheck('user_id', value):
self['user_id'] = value
else:
raise TypeError("cannot set user_id to " + repr(type(value)) + ", expecting: long")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment