How to use the faust.serializers 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.

github marcosschroh / python-schema-registry-client / tests / serializer / test_faust_serializer.py View on Github external
def test_dumps_load_with_register_codec(client, country_schema):
    payload = {"country": "Argenntina"}
    country_serializer = serializer.FaustSerializer(client, "test-country", country_schema)

    faust.serializers.codecs.register("country_serializer", country_serializer)

    class CountryRecord(faust.Record, serializer="country_serializer"):
        country: str

    country_record = CountryRecord(**payload)
    message_encoded = country_record.dumps()

    assert message_encoded
    assert len(message_encoded) > 5
    assert isinstance(message_encoded, bytes)

    message_decoded = CountryRecord.loads(message_encoded)

    assert message_decoded == country_record
github marcosschroh / python-schema-registry-client / tests / serializer / test_faust_serializer.py View on Github external
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()

    assert message_encoded