dataclass-and-union.py

gistfile1.txt

from dataclasses import dataclass, field
from typing import List, Optional, ClassVar, Type, Union

import marshmallow_dataclass
import marshmallow_union

from marshmallow import fields
from marshmallow.validate import Equal
from marshmallow_polyfield import PolyField
from marshmallow_toplevel import TopLevelSchema

@dataclass
class A:
    object_type: str = field(metadata={"validate": Equal("class_a")})
    field_x: str

@dataclass
class B:
    object_type: str = field(metadata={"validate": Equal("class_b")})
    field_x: Optional[str]
    field_y: str

@dataclass
class C:
    object_type: str = field(metadata={"validate": Equal("class_c")})
    field_z: str

ASchema = marshmallow_dataclass.class_schema(A)
BSchema = marshmallow_dataclass.class_schema(B)
CSchema = marshmallow_dataclass.class_schema(C)

class SourceStreamSchema(TopLevelSchema):
    _toplevel = marshmallow.fields.List(
        marshmallow_union.Union([
            fields.Nested(ASchema),
            fields.Nested(BSchema),
            fields.Nested(CSchema),
        ])
    )

source_stream = [
    {
        "object_type": "class_a",
        "field_x": "value_x",
    },
    {
        "object_type": "class_b",
        "field_x": "value_bx",
        "field_y": "value_by",
    },
    {
        "object_type": "class_c",
        "field_z": "value_z",
    },
    {
        "object_type": "class_b",
        "field_y": "value_by",
    },
]

output = SourceStreamSchema().load(source_stream)
output