python - 使从数据类创建的 faust.Record 与可选字段一起使用

Question

目前我正在编写一个 python 测试，所以我可以确保我的make_faust_record()方法有效。这就是它的样子：

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是使用 Google ProtocolBuffers 生成的，如您所见，第一个字段是id：

这是我得到的输出make_faust_record：

问题是，当我调用from时，.from_data()它抱怨第一个字段它不在有效负载（）中：@classmethodfaust.Recordid

我希望能够使用 INCOMPLETE PAYLOAD.from_data()对我的klass( ) 运行，因此只包含我在此 INCOMPLETE PAYLOAD 中提供的属性。faust.Recordfaust.Record

这是我的make_faust_record()逻辑：

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