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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
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 "skip_defaults" in kwargs:
            kwargs = dict(kwargs)
            skip_defaults = bool(kwargs.pop("skip_defaults"))
            if skip_defaults:
                kwargs["exclude_defaults"] = True
        return self.model_dump(mode="json", by_alias=True, exclude_none=True, **kwargs)

    def to_minimal_canonical_dict(self, **kwargs: Any) -> dict[str, Any]:
        """Convert instance to minimal canonical dictionary form.

        Minimal canonical form is intentionally compact:
        - excludes values equal to defaults
        - excludes ``None`` values
        - uses aliases for stable external contract
        """
        kwargs = dict(kwargs)
        kwargs.setdefault("exclude_defaults", True)
        kwargs.setdefault("exclude_none", True)
        kwargs.setdefault("by_alias", True)
        kwargs.setdefault("mode", "json")
        return self.model_dump(**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
78
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 "skip_defaults" in kwargs:
        kwargs = dict(kwargs)
        skip_defaults = bool(kwargs.pop("skip_defaults"))
        if skip_defaults:
            kwargs["exclude_defaults"] = True
    return self.model_dump(mode="json", by_alias=True, exclude_none=True, **kwargs)

to_minimal_canonical_dict(**kwargs)

Convert instance to minimal canonical dictionary form.

Minimal canonical form is intentionally compact: - excludes values equal to defaults - excludes None values - uses aliases for stable external contract

Source code in graflo/architecture/base.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def to_minimal_canonical_dict(self, **kwargs: Any) -> dict[str, Any]:
    """Convert instance to minimal canonical dictionary form.

    Minimal canonical form is intentionally compact:
    - excludes values equal to defaults
    - excludes ``None`` values
    - uses aliases for stable external contract
    """
    kwargs = dict(kwargs)
    kwargs.setdefault("exclude_defaults", True)
    kwargs.setdefault("exclude_none", True)
    kwargs.setdefault("by_alias", True)
    kwargs.setdefault("mode", "json")
    return self.model_dump(**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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
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)