avro模式的类型化类生成器
pyavro-gen的Python项目详细描述
皮亚夫罗发电机
python的标准avro实现是无类型的，在dicts上操作。
虽然这对于小项目、大项目（通常有数百个模式）很方便，
当然可以从记录构建期间强制模式的能力中获益。
这个库允许python avro用户使用特定的记录。
用法：
pip install pyavro-gen
现在，安装这个分支confluent_kafka
pip install -e git+https://github.com/confluentinc/confluent-kafka-python.git@serde#egg=confluent_kafka
pyavrogen.py -v \ -i my_schemas_dir \ -o jaumoavro \ -ie 'avsc'\ -b com.jaumo.schema \ -r com.jaumo.schema.rpc \ -t com.jaumo.schema.type
现在您可以导入您的类，如
fromjaumoavro.com.jaumo.schema.domain.userimportUpdatedu=Updated(...)
对于使用kafka avro serdes和fastavro的完整堆栈测试，运行
pyavrogen_test.py -m jaumoavro -d com.jaumo.schema.domain -n 100 -v
以编程方式生成和测试
fromavro_preprocessor.avro_pathsimportAvroPathsfrompyavro_gen.generatorimportAvroGeneratorfrompyavro_gen.pyavrogen_testimportdo_test_generated_classesgenerator=AvroGenerator(AvroPaths(input_path='myschemas/',output_path='avroclasses/',input_schema_file_extension='avsc',base_namespace='com.jaumo.schema',rpc_namespace='com.jaumo.schema.rpc',types_namespace='com.jaumo.schema.type',))generator.process()# Optionally, test the classes using fastavro:do_test_generated_classes(module='avroclasses/',domain_namespace='com.jaumo.schema.domain',number_of_cycles=100)
使用自定义类生成：
frompyavro_gen.generation_classesimportGENERATION_CLASSES,GenerationClassesTypefrompyavro_gen.codewriters.baseimportDecorator,ClassWriter,ExtensionfromtypingimportOptional# First define two custom ClassWritersclassRpcWriter(ClassWriter):def__init__(self,fully_qualified_name:str,doc:Optional[str]=None,prefix:Optional[str]=None):super().__init__(fully_qualified_name,doc,prefix)self.extensions=[Extension('abc.ABC')]classUndictifiableClassWriter(ClassWriter):def__init__(self,fully_qualified_name:str,doc:Optional[str]=None,prefix:Optional[str]=None):super().__init__(fully_qualified_name,doc,prefix)self.decorators=[Decorator('@type_checked_constructor()',ClassWriter('undictify.type_checked_constructor')),Decorator('@dataclass',ClassWriter('dataclasses.dataclass'))]# Then, register them in the GENERATION_CLASSES variableGENERATION_CLASSES[GenerationClassesType.RECORD_CLASS]=UndictifiableClassWriterGENERATION_CLASSES[GenerationClassesType.RPC_CLASS]=RpcWriter# Then, generate classes in module `avroclasses` as shown above.
欢迎加入QQ群-->： 979659372
