如何在Python数据类中为嵌套属性传递参数？

我尝试创建带有嵌套属性的Python dataclass实例，期望直接传递属于嵌套属性的参数时，该参数能自动替换对应嵌套属性的实际值。

尝试代码 #

from dataclasses import dataclass, field
from typing import Optional


@dataclass(kw_only=True)
class Attributes:
    correlation_id: Optional[str] = None
    routing_key: Optional[str] = None


@dataclass(kw_only=True)
class Metadata:
    workflow: Optional[str] = None
    workflow_state_id: Optional[str] = None
    attributes: Optional[Attributes] = field(default=None)

    def __post_init__(self):
        if self.attributes is None:
            self.attributes = Attributes()


@dataclass(kw_only=True)
class Message:
    version: Optional[int] = field(default=0)
    metadata: Optional[Metadata] = field(default=None)

    def __post_init__(self):
        if self.metadata is None:
            self.metadata = Metadata()


@dataclass(kw_only=True)
class Command(Message):
    ...


@dataclass(kw_only=True)
class SomeCommand(Command):
    name: str
    description: str


c1 = SomeCommand(name="awesome command", description="desc", routing_key='some key')
print(c1)

期望输出 #

SomeCommand(version=0, metadata=Metadata(workflow=None, workflow_state_id=None, attributes=Attributes(correlation_id=None, routing_key='some key')), name='awesome command', description='desc')

实际报错 #

TypeError: SomeCommand.__init__() got an unexpected keyword argument 'routing_key'

解决方案 #

可以实现该功能，核心思路是自定义__init__方法，将不属于当前dataclass的参数逐层传递给嵌套的子实例。以下是两种可行的修改方案：

方案一：逐层手动分离参数 #

from dataclasses import dataclass, field, fields
from typing import Optional


@dataclass(kw_only=True)
class Attributes:
    correlation_id: Optional[str] = None
    routing_key: Optional[str] = None


@dataclass(kw_only=True)
class Metadata:
    workflow: Optional[str] = None
    workflow_state_id: Optional[str] = None
    attributes: Optional[Attributes] = field(default=None)

    def __init__(self, **kwargs):
        # 分离当前类字段参数与Attributes的参数
        self_field_names = {f.name for f in fields(self)}
        meta_kwargs = {k: v for k, v in kwargs.items() if k in self_field_names}
        attr_kwargs = {k: v for k, v in kwargs.items() if k not in self_field_names}

        # 初始化当前类字段
        for name, value in meta_kwargs.items():
            setattr(self, name, value)
        
        # 初始化嵌套的Attributes实例
        self.attributes = Attributes(**attr_kwargs) if attr_kwargs else Attributes()


@dataclass(kw_only=True)
class Message:
    version: Optional[int] = field(default=0)
    metadata: Optional[Metadata] = field(default=None)

    def __init__(self, **kwargs):
        # 分离当前类字段参数与Metadata的参数
        self_field_names = {f.name for f in fields(self)}
        msg_kwargs = {k: v for k, v in kwargs.items() if k in self_field_names}
        meta_kwargs = {k: v for k, v in kwargs.items() if k not in self_field_names}

        # 初始化当前类字段
        for name, value in msg_kwargs.items():
            setattr(self, name, value)
        
        # 初始化嵌套的Metadata实例
        self.metadata = Metadata(**meta_kwargs) if meta_kwargs else Metadata()


@dataclass(kw_only=True)
class Command(Message):
    ...


@dataclass(kw_only=True)
class SomeCommand(Command):
    name: str
    description: str

    def __init__(self, **kwargs):
        # 分离当前类字段参数与父类的参数
        self_field_names = {f.name for f in fields(self)}
        cmd_kwargs = {k: v for k, v in kwargs.items() if k in self_field_names}
        parent_kwargs = {k: v for k, v in kwargs.items() if k not in self_field_names}

        # 先初始化父类
        super().__init__(**parent_kwargs)
        # 再初始化当前类字段
        for name, value in cmd_kwargs.items():
            setattr(self, name, value)


# 测试代码
c1 = SomeCommand(name="awesome command", description="desc", routing_key='some key', workflow='test_flow')
print(c1)

方案二：封装通用工具函数处理参数 #

如果嵌套层级较多，可封装工具函数减少重复代码：

from dataclasses import dataclass, field, fields
from typing import Optional, Dict, Type


def split_kwargs(cls: Type, kwargs: Dict) -> tuple[Dict, Dict]:
    """将传入参数拆分为当前类的字段参数和剩余参数"""
    cls_field_names = {f.name for f in fields(cls)}
    cls_kwargs = {k: v for k, v in kwargs.items() if k in cls_field_names}
    remaining_kwargs = {k: v for k, v in kwargs.items() if k not in cls_field_names}
    return cls_kwargs, remaining_kwargs


@dataclass(kw_only=True)
class Attributes:
    correlation_id: Optional[str] = None
    routing_key: Optional[str] = None


@dataclass(kw_only=True)
class Metadata:
    workflow: Optional[str] = None
    workflow_state_id: Optional[str] = None
    attributes: Optional[Attributes] = field(default=None)

    def __init__(self, **kwargs):
        meta_kwargs, attr_kwargs = split_kwargs(self.__class__, kwargs)
        for name, value in meta_kwargs.items():
            setattr(self, name, value)
        self.attributes = Attributes(**attr_kwargs) if attr_kwargs else Attributes()


@dataclass(kw_only=True)
class Message:
    version: Optional[int] = field(default=0)
    metadata: Optional[Metadata] = field(default=None)

    def __init__(self, **kwargs):
        msg_kwargs, meta_kwargs = split_kwargs(self.__class__, kwargs)
        for name, value in msg_kwargs.items():
            setattr(self, name, value)
        self.metadata = Metadata(**meta_kwargs) if meta_kwargs else Metadata()


@dataclass(kw_only=True)
class Command(Message):
    ...


@dataclass(kw_only=True)
class SomeCommand(Command):
    name: str
    description: str

    def __init__(self, **kwargs):
        cmd_kwargs, parent_kwargs = split_kwargs(self.__class__, kwargs)
        super().__init__(**parent_kwargs)
        for name, value in cmd_kwargs.items():
            setattr(self, name, value)


# 测试代码
c1 = SomeCommand(name="awesome command", description="desc", routing_key='some key', workflow_state_id='state_001')
print(c1)

代码说明 #

1. 通过fields()函数获取当前dataclass的所有字段名，将传入的kwargs拆分为当前类的参数和剩余参数。
2. 剩余参数会自动传递给下一层嵌套的dataclass实例，逐层处理直到最内层。
3. 重写__init__时需手动处理字段赋值，因为dataclass自动生成的__init__会被覆盖。

内容的提问来源于stack exchange，提问作者maudev