avro-gen3

import os import sys import tempfile from pathlib import Path from avrogen import write_schema_files # 1. Define a simple Avro schema avro_schema_json = ''' { "type": "record", "name": "User", "namespace": "com.example.app", "fields": [ {"name": "name", "type": "string"}, {"name": "favorite_number", "type": ["int", "null"], "default": null} ] } ''' # 2. Define an output directory for generated classes with tempfile.TemporaryDirectory() as tmpdir_name: output_dir = Path(tmpdir_name) print(f"Generated Avro classes will be written to: {output_dir}") # 3. Generate Python classes from the Avro schema write_schema_files(avro_schema_json, output_dir) # Add the output directory to sys.path to enable import sys.path.insert(0, str(output_dir)) try: # 4. Import the generated classes and reader # The Avro namespace 'com.example.app' translates to a path within the output_dir from com.example.app import User # Access the generated User class from avro.io import DatumWriter, DatumReader from avro.datafile import DataFileWriter, DataFileReader # 5. Create an instance of the generated class user_record = User(name="Alice", favorite_number=123) print(f"Created user record: {user_record}") print(f"User name: {user_record.name}, Favorite number: {user_record.favorite_number}") # 6. Serialize and deserialize using standard Avro tools with the generated schema/classes # Note: avro-gen3 wraps DatumReader but for DataFileWriter/Reader, you still use avro's types # For simpler examples, we might use the original avro library's DatumWriter/Reader directly # The main benefit of avro-gen3 is the type-hinted classes. # The generated classes are DictWrapper instances, compatible with standard Avro I/O output_file = output_dir / "users.avro" writer = DataFileWriter(open(output_file, "wb"), DatumWriter(), user_record.SCHEMA) writer.append(user_record._inner_dict) # avro-gen3 records are dict wrappers writer.close() reader = DataFileReader(open(output_file, "rb"), DatumReader()) for read_user_dict in reader: # When reading back, DatumReader returns dicts. You'd re-wrap if desired. read_user = User(**read_user_dict) print(f"Deserialized user: {read_user.name}, {read_user.favorite_number}") reader.close() finally: # Clean up sys.path sys.path.remove(str(output_dir))