How to use the faust.serializers.codecs.register function in faust
To help you get started, we’ve selected a few faust examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
marcosschroh / python-schema-registry-client / tests / serializer / test_faust_serializer.py View on Github 
def test_nested_schema_with_register_codec(client):
nested_schema = schema.AvroSchema(data_gen.NESTED_SCHENA)
order_schema = schema.AvroSchema(data_gen.ORDER_SCHENA)
customer_serializer = serializer.FaustSerializer(client, "test-nested-schema", nested_schema)
order_serializer = serializer.FaustSerializer(client, "test-order-schema", order_schema)
faust.serializers.codecs.register("customer_serializer", customer_serializer)
faust.serializers.codecs.register("order_serializer", order_serializer)
class Order(faust.Record, serializer="order_serializer"):
uid: int
class Customer(faust.Record, serializer="customer_serializer"):
name: str
uid: int
order: Order
payload = data_gen.create_nested_schema()
customer = Customer(**payload)
message_encoded = customer.dumps()
