Schema resources

schema

SchemaProvider

Schema-scoped provider — safe to import inside a Snowpark stored procedure.

Deploys only objects that can exist within a client schema: tables, views, procedures, stages, secrets (client credentials), tasks, streams, and so on. Never creates account-level objects (warehouses, roles, EAIs).

Intended for use in PROVISION_CLIENT() running via EXECUTE AS OWNER inside Snowflake. Import path kept lightweight for ARTIFACT_REPOSITORY.

Example

.. code-block:: python

# ARTIFACT_REPOSITORY_PACKAGES = ('pinky-provider[schema]')
from pinky_provider.resources.schema import SchemaProvider

def provision_client(session, team_id, product_db, client_id):
    provider = SchemaProvider(session)
    provider.apply(
        source=f"{product_db}.PUBLIC.GIT_REPO:/manifest_client/",
        target=f"{product_db}.{client_id}",
        vars={"client_id": client_id},
    )

Parameters:

Name	Type	Description	Default
session	Any	Active Snowpark Session running EXECUTE AS OWNER.	required

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

class SchemaProvider:
    """Schema-scoped provider — safe to import inside a Snowpark stored procedure.

    Deploys only objects that can exist within a client schema: tables, views,
    procedures, stages, secrets (client credentials), tasks, streams, and so on.
    Never creates account-level objects (warehouses, roles, EAIs).

    Intended for use in ``PROVISION_CLIENT()`` running via ``EXECUTE AS OWNER``
    inside Snowflake. Import path kept lightweight for ``ARTIFACT_REPOSITORY``.

    Example:
        .. code-block:: python

            # ARTIFACT_REPOSITORY_PACKAGES = ('pinky-provider[schema]')
            from pinky_provider.resources.schema import SchemaProvider

            def provision_client(session, team_id, product_db, client_id):
                provider = SchemaProvider(session)
                provider.apply(
                    source=f"{product_db}.PUBLIC.GIT_REPO:/manifest_client/",
                    target=f"{product_db}.{client_id}",
                    vars={"client_id": client_id},
                )

    Args:
        session: Active Snowpark ``Session`` running ``EXECUTE AS OWNER``.
    """

    def __init__(self, session: Any) -> None:
        self._session = session

    def plan(
        self,
        source: str,
        target: str,
        vars: dict[str, str] | None = None,
    ) -> list[Any]:
        """Compute changes without applying them.

        Args:
            source: ``manifest.yml`` path or Snowflake Git repo URL.
            target: Target schema as ``DATABASE.SCHEMA``.
            vars: Jinja2 variables injected into the manifest.

        Returns:
            Ordered list of :class:`~pinky_provider.core.diff.ResourceDiff`.
        """
        raise NotImplementedError("SchemaProvider is not implemented yet")

    def apply(
        self,
        source: str,
        target: str,
        vars: dict[str, str] | None = None,
    ) -> None:
        """Apply the manifest to the target schema.

        Args:
            source: ``manifest.yml`` path or Snowflake Git repo URL.
            target: Target schema as ``DATABASE.SCHEMA``.
            vars: Jinja2 variables injected into the manifest.
        """
        ...

apply(source, target, vars=None)

Apply the manifest to the target schema.

Parameters:

Name	Type	Description	Default
source	str	manifest.yml path or Snowflake Git repo URL.	required
target	str	Target schema as DATABASE.SCHEMA.	required
vars	dict[str, str] | None	Jinja2 variables injected into the manifest.	None

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

def apply(
    self,
    source: str,
    target: str,
    vars: dict[str, str] | None = None,
) -> None:
    """Apply the manifest to the target schema.

    Args:
        source: ``manifest.yml`` path or Snowflake Git repo URL.
        target: Target schema as ``DATABASE.SCHEMA``.
        vars: Jinja2 variables injected into the manifest.
    """
    ...

plan(source, target, vars=None)

Compute changes without applying them.

Parameters:

Name	Type	Description	Default
source	str	manifest.yml path or Snowflake Git repo URL.	required
target	str	Target schema as DATABASE.SCHEMA.	required
vars	dict[str, str] | None	Jinja2 variables injected into the manifest.	None

Returns:

Type	Description
list[Any]	Ordered list of :class:~pinky_provider.core.diff.ResourceDiff.

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

def plan(
    self,
    source: str,
    target: str,
    vars: dict[str, str] | None = None,
) -> list[Any]:
    """Compute changes without applying them.

    Args:
        source: ``manifest.yml`` path or Snowflake Git repo URL.
        target: Target schema as ``DATABASE.SCHEMA``.
        vars: Jinja2 variables injected into the manifest.

    Returns:
        Ordered list of :class:`~pinky_provider.core.diff.ResourceDiff`.
    """
    raise NotImplementedError("SchemaProvider is not implemented yet")

table

iceberg_table

view

dynamic_table

procedure

user_defined_function

task

stage

stream

event_table

alert

streamlit

tag

secret

storage_lifecycle_policy