Account resources

account

AccountProvider

Account-scoped provider — deploys account-level objects (warehouses, roles, etc.).

Never imported inside a stored procedure. Used only by the CLI in CI/CD context.

Parameters:

Name	Type	Description	Default
session	Any	Active Snowpark Session with ACCOUNTADMIN or equivalent role.	required

Source code in src/pinky_provider/resources/account/__init__.py

class AccountProvider:
    """Account-scoped provider — deploys account-level objects (warehouses, roles, etc.).

    Never imported inside a stored procedure. Used only by the CLI in CI/CD context.

    Args:
        session: Active Snowpark ``Session`` with ``ACCOUNTADMIN`` or equivalent role.
    """

    def __init__(self, session: Any) -> None:
        from snowflake.core import Root

        self._session = session
        self._root = Root(session)

    def _build(self, spec: ResourceSpec) -> BaseResource:
        """Construct a validated resource model from a manifest spec.

        ``_``-prefixed YAML keys (``_depends_on``, ``_tags``) are stripped before validation and
        re-attached as private attributes, so ``extra="forbid"`` only sees Snowflake fields.
        """
        cls = RESOURCES.get(spec.resource_type)
        if cls is None:
            raise ValueError(f"Unsupported account resource type: {spec.resource_type!r}")
        meta = {k: v for k, v in spec.data.items() if k.startswith("_")}
        fields = {k: v for k, v in spec.data.items() if not k.startswith("_")}
        resource = cls(**fields)
        resource._depends_on = meta.get("_depends_on", [])
        resource._tags = meta.get("_tags", {})
        return resource

    def _fetch(self, resource: BaseResource) -> dict[str, Any] | None:
        """Fetch the current Snowflake state, or ``None`` if the object does not exist."""
        from snowflake.core.exceptions import NotFoundError

        try:
            current = resource.collection(self._root)[resource.name].fetch().to_dict()
        except NotFoundError:
            return None
        return dict(current)

    def _diff(self, spec: ResourceSpec, resource: BaseResource) -> ResourceDiff:
        desired = resource.to_sdk_dict()  # exactly what Snowflake will receive
        actual = self._fetch(resource)
        return compute_diff(
            spec.resource_type,
            resource.name,
            desired,
            actual,
            normalizers=type(resource)._field_normalizers,
        )

    def plan(self, manifest: Manifest) -> list[ResourceDiff]:
        """Compute changes without applying them."""
        return [self._diff(spec, self._build(spec)) for spec in manifest.resources]

    def apply(self, manifest: Manifest) -> list[ResourceDiff]:
        """Apply the manifest at account scope — creates or alters each declared object.

        Returns the diff list actually executed, so the CLI can render what happened.
        """
        applied: list[ResourceDiff] = []
        for spec in manifest.resources:
            resource = self._build(spec)
            diff = self._diff(spec, resource)
            if diff.action in (Action.CREATE, Action.ALTER):
                inject_query_tag(
                    self._session,
                    repo=manifest.root_dir.name,
                    schema=manifest.scope,
                    resource_type=spec.resource_type,
                    name=resource.name,
                )
                resource.collection(self._root)[resource.name].create_or_alter(resource.to_sdk())
            applied.append(diff)
        return applied

apply(manifest)

Apply the manifest at account scope — creates or alters each declared object.

Returns the diff list actually executed, so the CLI can render what happened.

Source code in src/pinky_provider/resources/account/__init__.py

def apply(self, manifest: Manifest) -> list[ResourceDiff]:
    """Apply the manifest at account scope — creates or alters each declared object.

    Returns the diff list actually executed, so the CLI can render what happened.
    """
    applied: list[ResourceDiff] = []
    for spec in manifest.resources:
        resource = self._build(spec)
        diff = self._diff(spec, resource)
        if diff.action in (Action.CREATE, Action.ALTER):
            inject_query_tag(
                self._session,
                repo=manifest.root_dir.name,
                schema=manifest.scope,
                resource_type=spec.resource_type,
                name=resource.name,
            )
            resource.collection(self._root)[resource.name].create_or_alter(resource.to_sdk())
        applied.append(diff)
    return applied

plan(manifest)

Compute changes without applying them.

Source code in src/pinky_provider/resources/account/__init__.py

def plan(self, manifest: Manifest) -> list[ResourceDiff]:
    """Compute changes without applying them."""
    return [self._diff(spec, self._build(spec)) for spec in manifest.resources]

database

warehouse

Warehouse

Bases: BaseResource

Account-scoped warehouse — fields introspected from snowflake.core.warehouse.Warehouse.

Deployed via Core API create_or_alter (ADR-0005) — no SQL fallback needed. Two SDK warts are curated for a clean YAML surface (ADR-0015): warehouse_size is exposed as :class:WarehouseSize (autocompleted, typo-checked) and auto_resume as a real bool.

Source code in src/pinky_provider/resources/account/warehouse.py

class Warehouse(BaseResource):
    """Account-scoped warehouse — fields introspected from ``snowflake.core.warehouse.Warehouse``.

    Deployed via Core API ``create_or_alter`` (ADR-0005) — no SQL fallback needed. Two SDK warts are
    curated for a clean YAML surface (ADR-0015): ``warehouse_size`` is exposed as :class:`WarehouseSize`
    (autocompleted, typo-checked) and ``auto_resume`` as a real ``bool``.
    """

    _sdk_class: ClassVar[type] = _SDKWarehouse
    _field_overrides: ClassVar[dict[str, tuple[Any, Any]]] = {
        # soft enum: autocomplete the known sizes, but let an unknown/future size through
        "warehouse_size": (WarehouseSize | str, str),
        "auto_resume": (bool, _bool_to_sdk),
    }
    _field_normalizers: ClassVar[dict[str, Any]] = {"warehouse_size": _canon_size}

    def collection(self, root: Any) -> Any:
        """Account-level warehouse collection."""
        return root.warehouses

collection(root)

Account-level warehouse collection.

Source code in src/pinky_provider/resources/account/warehouse.py

def collection(self, root: Any) -> Any:
    """Account-level warehouse collection."""
    return root.warehouses

WarehouseSize

Bases: StrEnum

Curated warehouse sizes — drives YAML autocomplete (ADR-0015).

Snowflake's SDK types warehouse_size as a bare str; exposing this enum gives the editor a value dropdown. Used as a soft enum (WarehouseSize | str): unknown values still pass — forward-compatible if Snowflake adds a tier before we do — and are validated by Snowflake at apply. Extend by one line to get autocomplete for a new tier.

Source code in src/pinky_provider/resources/account/warehouse.py

class WarehouseSize(StrEnum):
    """Curated warehouse sizes — drives YAML autocomplete (ADR-0015).

    Snowflake's SDK types ``warehouse_size`` as a bare ``str``; exposing this enum gives the editor
    a value dropdown. Used as a **soft** enum (``WarehouseSize | str``): unknown values still pass —
    forward-compatible if Snowflake adds a tier before we do — and are validated by Snowflake at
    ``apply``. Extend by one line to get autocomplete for a new tier.
    """

    XSMALL = "XSMALL"
    SMALL = "SMALL"
    MEDIUM = "MEDIUM"
    LARGE = "LARGE"
    XLARGE = "XLARGE"
    XXLARGE = "XXLARGE"
    XXXLARGE = "XXXLARGE"
    X4LARGE = "X4LARGE"
    X5LARGE = "X5LARGE"
    X6LARGE = "X6LARGE"

role

user

network_policy

network_rule

api_integration

external_access_integration

external_volume

compute_pool

notification_integration

tag

secret