Skip to content

graflo.architecture.base

Base model for Graflo configuration classes with YAML support.

ConfigBaseModel

Bases: BaseModel

Base model for all Graflo configuration classes.

Provides YAML serialization/deserialization and standard configuration for all Pydantic models in the system.

Source code in graflo/architecture/base.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class ConfigBaseModel(BaseModel):
    """Base model for all Graflo configuration classes.

    Provides YAML serialization/deserialization and standard configuration
    for all Pydantic models in the system.
    """

    model_config = ConfigDict(
        populate_by_name=True,
        extra="forbid",
        use_enum_values=True,
        validate_assignment=True,
    )

    @classmethod
    def from_yaml(cls, path: str) -> Self:
        """Load a single instance from a YAML file."""
        with open(path) as f:
            data = yaml.safe_load(f)
        return cls.model_validate(data)

    @classmethod
    def from_yaml_list(cls, path: str) -> list[Self]:
        """Load a list of instances from a YAML file."""
        with open(path) as f:
            data = yaml.safe_load(f)
        if not isinstance(data, list):
            raise ValueError(f"Expected list in YAML file, got {type(data)}")
        return [cls.model_validate(item) for item in data]

    @classmethod
    def from_dict(cls, data: dict[str, Any] | list[Any]) -> Self:
        """Load from a dictionary (or list for root model)."""
        return cls.model_validate(data)

    def to_yaml(self, path: str, **kwargs: Any) -> None:
        """Save instance to a YAML file."""
        with open(path, "w") as f:
            yaml.safe_dump(
                self.model_dump(by_alias=True, exclude_none=True),
                f,
                default_flow_style=False,
                sort_keys=False,
                **kwargs,
            )

    def to_yaml_str(self, **kwargs: Any) -> str:
        """Convert instance to a YAML string."""
        return yaml.safe_dump(
            self.model_dump(by_alias=True, exclude_none=True),
            default_flow_style=False,
            sort_keys=False,
            **kwargs,
        )

    def to_dict(self, **kwargs: Any) -> dict[str, Any]:
        """Convert instance to a dictionary.

        Supports skip_defaults=True (mapped to exclude_defaults) for backward
        compatibility with dataclass-wizard style APIs.
        """
        if kwargs.get("skip_defaults"):
            kwargs = dict(kwargs)
            kwargs.pop("skip_defaults", None)
            kwargs["exclude_defaults"] = True
        return self.model_dump(by_alias=True, exclude_none=True, **kwargs)

    def update(self, other: Self) -> None:
        """Update this instance with values from another instance of the same type.

        Performs in-place merge: lists are concatenated, dicts/sets are merged,
        nested ConfigBaseModel instances are updated recursively. None values
        in other do not overwrite existing values.

        Args:
            other: Another instance of the same type to copy from

        Raises:
            TypeError: If other is not an instance of the same type
        """
        if type(other) is not type(self):
            raise TypeError(
                f"Expected {type(self).__name__} instance, got {type(other).__name__}"
            )
        for name in type(self).model_fields:
            current = getattr(self, name)
            other_val = getattr(other, name)
            if other_val is None:
                continue
            if isinstance(other_val, list):
                setattr(self, name, current + deepcopy(other_val))
            elif isinstance(other_val, set):
                setattr(self, name, current | deepcopy(other_val))
            elif isinstance(other_val, dict):
                setattr(self, name, {**current, **deepcopy(other_val)})
            elif isinstance(other_val, ConfigBaseModel):
                if current is not None:
                    current.update(other_val)
                else:
                    setattr(self, name, deepcopy(other_val))
            else:
                if current is None:
                    setattr(self, name, other_val)

from_dict(data) classmethod

Load from a dictionary (or list for root model).

Source code in graflo/architecture/base.py 
42
43
44
45
@classmethod
def from_dict(cls, data: dict[str, Any] | list[Any]) -> Self:
    """Load from a dictionary (or list for root model)."""
    return cls.model_validate(data)

from_yaml(path) classmethod

Load a single instance from a YAML file.

Source code in graflo/architecture/base.py 
26
27
28
29
30
31
@classmethod
def from_yaml(cls, path: str) -> Self:
    """Load a single instance from a YAML file."""
    with open(path) as f:
        data = yaml.safe_load(f)
    return cls.model_validate(data)

from_yaml_list(path) classmethod

Load a list of instances from a YAML file.

Source code in graflo/architecture/base.py 
33
34
35
36
37
38
39
40
@classmethod
def from_yaml_list(cls, path: str) -> list[Self]:
    """Load a list of instances from a YAML file."""
    with open(path) as f:
        data = yaml.safe_load(f)
    if not isinstance(data, list):
        raise ValueError(f"Expected list in YAML file, got {type(data)}")
    return [cls.model_validate(item) for item in data]

to_dict(**kwargs)

Convert instance to a dictionary.

Supports skip_defaults=True (mapped to exclude_defaults) for backward compatibility with dataclass-wizard style APIs.

Source code in graflo/architecture/base.py 
67
68
69
70
71
72
73
74
75
76
77
def to_dict(self, **kwargs: Any) -> dict[str, Any]:
    """Convert instance to a dictionary.

    Supports skip_defaults=True (mapped to exclude_defaults) for backward
    compatibility with dataclass-wizard style APIs.
    """
    if kwargs.get("skip_defaults"):
        kwargs = dict(kwargs)
        kwargs.pop("skip_defaults", None)
        kwargs["exclude_defaults"] = True
    return self.model_dump(by_alias=True, exclude_none=True, **kwargs)

to_yaml(path, **kwargs)

Save instance to a YAML file.

Source code in graflo/architecture/base.py 
47
48
49
50
51
52
53
54
55
56
def to_yaml(self, path: str, **kwargs: Any) -> None:
    """Save instance to a YAML file."""
    with open(path, "w") as f:
        yaml.safe_dump(
            self.model_dump(by_alias=True, exclude_none=True),
            f,
            default_flow_style=False,
            sort_keys=False,
            **kwargs,
        )

to_yaml_str(**kwargs)

Convert instance to a YAML string.

Source code in graflo/architecture/base.py 
58
59
60
61
62
63
64
65
def to_yaml_str(self, **kwargs: Any) -> str:
    """Convert instance to a YAML string."""
    return yaml.safe_dump(
        self.model_dump(by_alias=True, exclude_none=True),
        default_flow_style=False,
        sort_keys=False,
        **kwargs,
    )

update(other)

Update this instance with values from another instance of the same type.

Performs in-place merge: lists are concatenated, dicts/sets are merged, nested ConfigBaseModel instances are updated recursively. None values in other do not overwrite existing values.

Parameters:

Name Type Description Default
other Self

Another instance of the same type to copy from

 required

Raises:

Type Description
TypeError

If other is not an instance of the same type
Source code in graflo/architecture/base.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def update(self, other: Self) -> None:
    """Update this instance with values from another instance of the same type.

    Performs in-place merge: lists are concatenated, dicts/sets are merged,
    nested ConfigBaseModel instances are updated recursively. None values
    in other do not overwrite existing values.

    Args:
        other: Another instance of the same type to copy from

    Raises:
        TypeError: If other is not an instance of the same type
    """
    if type(other) is not type(self):
        raise TypeError(
            f"Expected {type(self).__name__} instance, got {type(other).__name__}"
        )
    for name in type(self).model_fields:
        current = getattr(self, name)
        other_val = getattr(other, name)
        if other_val is None:
            continue
        if isinstance(other_val, list):
            setattr(self, name, current + deepcopy(other_val))
        elif isinstance(other_val, set):
            setattr(self, name, current | deepcopy(other_val))
        elif isinstance(other_val, dict):
            setattr(self, name, {**current, **deepcopy(other_val)})
        elif isinstance(other_val, ConfigBaseModel):
            if current is not None:
                current.update(other_val)
            else:
                setattr(self, name, deepcopy(other_val))
        else:
            if current is None:
                setattr(self, name, other_val)