Make faust.Record created from dataclass works with optional fields

267 Views Asked by At

Currently i'am writting a python test so I can assure my make_faust_record() method works. This is how is looks like:

def test_order_supports_optional_fields_records():
    klass = make_faust_record(Order)
    payload = {"pos": {"hub": {"name": "Kitchen A"}}}
    instance = klass.from_data(payload)
    assert instance.pos.hub.name == "Kitchen A"
    assert instance.pos.brand is None

Order was generated using Google ProtocolBuffers, and as you can see, the first field is id:

enter image description here

This is the output i got from make_faust_record:

enter image description here

The problem is that when i call .from_data(), which is a @classmethod from faust.Record, it complains about the first field it which is not into the payload (id):

enter image description here

I wanna be able to run .from_data() against my klass (faust.Record), using a INCOMPLETE PAYLOAD and have as a result a faust.Record only contain the properties i provided in this INCOMPLETE PAYLOAD.

This is my make_faust_record() logic:

import abc
import inspect
from typing import Optional, Type

import betterproto
import faust

GENERATED_SUFFIX = "__FaustRecord_Auto"


def _import_relative_class(module: str, klass_name: str):
    resolved_import = __import__(module, fromlist=[klass_name])
    klass = getattr(resolved_import, klass_name)
    return klass


def _is_record(attype: Type):
    return (
        inspect.isclass(attype)
        and isinstance(attype, betterproto.Message)
        or isinstance(attype, abc.ABCMeta)
    )


def _build_record_annotations(klass: Type):
    annotations = {}
    for atname, attype in klass.__annotations__.items():
        final_attype = attype
        if _is_record(attype):
            final_attype = make_faust_record(attype)
        elif isinstance(attype, str):
            subklass = _import_relative_class(klass.__module__, attype)
            if _is_record(subklass):
                final_attype = make_faust_record(subklass)
            else:
                final_attype = str
        annotations[atname] = Optional[final_attype]

    return annotations


def make_faust_record(klass: Type):
    type_name = f"{klass.__name__}{GENERATED_SUFFIX}"
    record_type = type(type_name, (faust.Record, klass), {})
    record_type.__annotations__ = _build_record_annotations(klass)
    record_type._init_subclass()

    return record_type
0

There are 0 best solutions below