Skip to content

base

BaseResource

Bases: BaseModel

Base Pydantic model for all pinky-provider resource types.

Each resource subclass represents one Snowflake object as declared in a .yml file. Field definitions are derived at class-creation time from _sdk_class via :func:_sdk_fields — no manual mapping (ADR-0001).

Custom YAML keys prefixed with _ (_depends_on, _tags) are carried as private attributes and excluded from :meth:to_sdk — they drive the provider, not the DDL.

Attributes:

Name Type Description
name str

The Snowflake object name (maps to the YAML name key).
_depends_on list[str]

Ordered deployment dependencies, expressed as <type>/<name> paths.
_tags dict[str, str]

Tag name → value, applied via ALTER … SET TAG after create.
Source code in src/pinky_provider/resources/base.py 
 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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
class BaseResource(BaseModel, metaclass=_ResourceMeta):
    """Base Pydantic model for all pinky-provider resource types.

    Each resource subclass represents one Snowflake object as declared in a ``.yml`` file.
    Field definitions are derived at class-creation time from ``_sdk_class`` via
    :func:`_sdk_fields` — no manual mapping (ADR-0001).

    Custom YAML keys prefixed with ``_`` (``_depends_on``, ``_tags``) are carried as private
    attributes and excluded from :meth:`to_sdk` — they drive the provider, not the DDL.

    Attributes:
        name: The Snowflake object name (maps to the YAML ``name`` key).
        _depends_on: Ordered deployment dependencies, expressed as ``<type>/<name>`` paths.
        _tags: Tag name → value, applied via ``ALTER … SET TAG`` after create.
    """

    # extra=forbid → catch typos / fields absent from the SDK at validation time.
    # arbitrary_types_allowed → some snowflake.core fields are typed with non-pydantic classes;
    # accept them as-is rather than failing the model build.
    model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True)

    _sdk_class: ClassVar[type[Any] | None] = None
    # field → (clean Pydantic type/annotation, caster back to the SDK form). See _ResourceMeta
    # (ADR-0015). The type may be a union (e.g. ``WarehouseSize | str``) for a soft enum.
    _field_overrides: ClassVar[dict[str, tuple[Any, Callable[[Any], Any]]]] = {}
    # field → comparison normaliser for the diff engine (Snowflake echoes display forms, e.g.
    # warehouse_size XSMALL → "X-Small"). Default comparison is case-insensitive.
    _field_normalizers: ClassVar[dict[str, Callable[[Any], Any]]] = {}

    name: str
    _depends_on: list[str] = PrivateAttr(default_factory=list)
    _tags: dict[str, str] = PrivateAttr(default_factory=dict)

    _MANAGED_COMMENT: ClassVar[str] = "Managed by pinky-provider"

    def to_sdk_dict(self) -> dict[str, Any]:
        """The exact field payload sent to the SDK — overrides cast back to their SDK form.

        Drops ``_``-prefixed keys and unset fields, applies each ``_field_overrides`` caster (e.g.
        ``bool`` → ``"true"``, enum → its string value), and injects the managed ``comment``. The
        diff engine compares against this, so plan/apply see exactly what Snowflake will receive.
        """
        if self._sdk_class is None:  # pragma: no cover — guarded by metaclass usage
            raise TypeError(f"{type(self).__name__} declares no _sdk_class")

        overrides = type(self)._field_overrides
        data: dict[str, Any] = {}
        for key, value in self.model_dump(exclude_none=True).items():
            if key.startswith("_"):
                continue
            data[key] = overrides[key][1](value) if key in overrides else value
        if "comment" in _sdk_fields(self._sdk_class) and "comment" not in data:
            data["comment"] = self._MANAGED_COMMENT
        return data

    def to_sdk(self) -> Any:
        """Build the ``snowflake.core`` SDK object from :meth:`to_sdk_dict`."""
        payload = self.to_sdk_dict()  # raises if _sdk_class is None
        assert self._sdk_class is not None  # narrowing for the type checker
        return self._sdk_class(**payload)

    def collection(self, root: Any) -> Any:
        """Return the ``snowflake.core`` collection this resource lives in (e.g. ``root.warehouses``).

        Overridden by each resource type. Used by the scope providers to fetch / create / alter.
        """
        raise NotImplementedError(f"{type(self).__name__} does not implement collection()")

collection(root)

Return the snowflake.core collection this resource lives in (e.g. root.warehouses).

Overridden by each resource type. Used by the scope providers to fetch / create / alter.

Source code in src/pinky_provider/resources/base.py 
151
152
153
154
155
156
def collection(self, root: Any) -> Any:
    """Return the ``snowflake.core`` collection this resource lives in (e.g. ``root.warehouses``).

    Overridden by each resource type. Used by the scope providers to fetch / create / alter.
    """
    raise NotImplementedError(f"{type(self).__name__} does not implement collection()")

to_sdk()

Build the snowflake.core SDK object from :meth:to_sdk_dict.

Source code in src/pinky_provider/resources/base.py 
145
146
147
148
149
def to_sdk(self) -> Any:
    """Build the ``snowflake.core`` SDK object from :meth:`to_sdk_dict`."""
    payload = self.to_sdk_dict()  # raises if _sdk_class is None
    assert self._sdk_class is not None  # narrowing for the type checker
    return self._sdk_class(**payload)

to_sdk_dict()

The exact field payload sent to the SDK — overrides cast back to their SDK form.

Drops _-prefixed keys and unset fields, applies each _field_overrides caster (e.g. bool → "true", enum → its string value), and injects the managed comment. The diff engine compares against this, so plan/apply see exactly what Snowflake will receive.

Source code in src/pinky_provider/resources/base.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def to_sdk_dict(self) -> dict[str, Any]:
    """The exact field payload sent to the SDK — overrides cast back to their SDK form.

    Drops ``_``-prefixed keys and unset fields, applies each ``_field_overrides`` caster (e.g.
    ``bool`` → ``"true"``, enum → its string value), and injects the managed ``comment``. The
    diff engine compares against this, so plan/apply see exactly what Snowflake will receive.
    """
    if self._sdk_class is None:  # pragma: no cover — guarded by metaclass usage
        raise TypeError(f"{type(self).__name__} declares no _sdk_class")

    overrides = type(self)._field_overrides
    data: dict[str, Any] = {}
    for key, value in self.model_dump(exclude_none=True).items():
        if key.startswith("_"):
            continue
        data[key] = overrides[key][1](value) if key in overrides else value
    if "comment" in _sdk_fields(self._sdk_class) and "comment" not in data:
        data["comment"] = self._MANAGED_COMMENT
    return data