sqlmesh.core.config.base

  1from __future__ import annotations
  2
  3import typing as t
  4from enum import Enum, auto
  5
  6from sqlmesh.utils.errors import ConfigError
  7from sqlmesh.utils.pydantic import PydanticModel
  8
  9T = t.TypeVar("T", bound="BaseConfig")
 10
 11
 12class UpdateStrategy(Enum):
 13    """Supported strategies for adding new config to existing config"""
 14
 15    REPLACE = auto()  # Replace with new value
 16    EXTEND = auto()  # Extend list to existing list
 17    KEY_UPDATE = auto()  # Update dict key value with new dict key value
 18    KEY_EXTEND = auto()  # Extend dict key value to existing dict key value
 19    IMMUTABLE = auto()  # Raise if a key tries to change this value
 20    NESTED_UPDATE = auto()  # Recursively updates the nested config
 21
 22
 23def update_field(
 24    old: t.Optional[t.Any],
 25    new: t.Any,
 26    update_strategy: t.Optional[UpdateStrategy] = None,
 27) -> t.Any:
 28    """
 29    Update config field with new config value
 30
 31    Args:
 32        old: The existing config value
 33        new: The new config value
 34        update_strategy: The strategy to use when updating the field
 35
 36
 37        The updated field
 38    """
 39    if not old:
 40        return new
 41
 42    update_strategy = update_strategy or UpdateStrategy.REPLACE
 43
 44    if update_strategy == UpdateStrategy.IMMUTABLE:
 45        raise ConfigError(f"Cannot modify property: {old}.")
 46
 47    if update_strategy == UpdateStrategy.REPLACE:
 48        return new
 49    if update_strategy == UpdateStrategy.EXTEND:
 50        if not isinstance(old, list) or not isinstance(new, list):
 51            raise ConfigError("EXTEND behavior requires list field.")
 52
 53        return old + new
 54    if update_strategy == UpdateStrategy.KEY_UPDATE:
 55        if not isinstance(old, dict) or not isinstance(new, dict):
 56            raise ConfigError("KEY_UPDATE behavior requires dictionary field.")
 57
 58        combined = old.copy()
 59        combined.update(new)
 60        return combined
 61    if update_strategy == UpdateStrategy.KEY_EXTEND:
 62        if not isinstance(old, dict) or not isinstance(new, dict):
 63            raise ConfigError("KEY_EXTEND behavior requires dictionary field.")
 64
 65        combined = old.copy()
 66        for key, value in new.items():
 67            if not isinstance(value, list):
 68                raise ConfigError("KEY_EXTEND behavior requires list values in dictionary.")
 69
 70            old_value = combined.get(key)
 71            if old_value:
 72                if not isinstance(old_value, list):
 73                    raise ConfigError("KEY_EXTEND behavior requires list values in dictionary.")
 74
 75                combined[key] = old_value + value
 76            else:
 77                combined[key] = value
 78
 79        return combined
 80    if update_strategy == UpdateStrategy.NESTED_UPDATE:
 81        if not isinstance(old, BaseConfig):
 82            raise ConfigError(
 83                f"NESTED_UPDATE behavior requires a config object. {type(old)} was given instead."
 84            )
 85
 86        if type(new) != type(old):
 87            raise ConfigError(
 88                "NESTED_UPDATE behavior requires both values to have the same type. "
 89                f"{type(old)} and {type(new)} were given instead."
 90            )
 91
 92        return old.update_with(new)
 93
 94    raise ConfigError(f"Unknown update strategy {update_strategy}.")
 95
 96
 97class BaseConfig(PydanticModel):
 98    """Base configuration functionality for configuration classes."""
 99
100    _FIELD_UPDATE_STRATEGY: t.ClassVar[t.Dict[str, UpdateStrategy]] = {}
101
102    def update_with(self: T, other: t.Union[t.Dict[str, t.Any], T]) -> T:
103        """Updates this instance's fields with the passed in config fields and returns a new instance.
104
105        Args:
106            other: Other configuration.
107
108        Returns:
109            New instance updated with the passed in config fields
110        """
111        if isinstance(other, dict):
112            other = self.__class__(**other)
113
114        updated_fields = {}
115
116        for field in other.fields_set:
117            if field in self.all_field_infos():
118                updated_fields[field] = update_field(
119                    getattr(self, field),
120                    getattr(other, field),
121                    self._FIELD_UPDATE_STRATEGY.get(field),
122                )
123            else:
124                updated_fields[field] = getattr(other, field)
125
126        # Assign each field to trigger assignment validators
127        updated = self.copy()
128        for field, value in updated_fields.items():
129            setattr(updated, field, value)
130
131        return updated