Skip to content

Commit 6171dfb

Browse files
eric-thomas-dagsterclaude
eric-thomas-dagster
and
claude
committed
feat(snowflake): 6 single-entity components — task/proc/DT actions + completion sensor + OpenFlow external+sensor
Single-entity counterparts to snowflake_workspace's auto-discovered assets, plus the OpenFlow external-asset + sensor pair Eric called out. New asset components (assets/data-warehouse/): - snowflake_task_execute_asset — EXECUTE TASK on one named task; pulls last TASK_HISTORY entry into metadata. Auth via password / PAT / keypair env vars. - snowflake_stored_procedure_call_asset — CALL <db>.<schema>.<proc>(args) with auto-quoted string args + inlined numeric/boolean args. - snowflake_dynamic_table_refresh_asset — ALTER DYNAMIC TABLE ... REFRESH on one named DT; surfaces scheduling_state + last_refresh_status. New sensor components (sensors/): - snowflake_task_completion_sensor — polls TASK_HISTORY for new SUCCESS rows. Three modes via asset_key + job_name fields: observe-only (AssetMaterialization), trigger-only (RunRequest), or both. Cursors on most-recent query_id for dedup. - snowflake_openflow_status_sensor — polls SNOWFLAKE.TELEMETRY.EVENTS for openflow_metric activity on a named flow. Same three modes. Cursors on MAX(TIMESTAMP) for the flow. New external-asset component (external_assets/): - external_snowflake_openflow_flow — declare-only AssetSpec for an OpenFlow process group. OpenFlow runs the flow on its own infra (BYOC EKS or Snowflake-managed); the openflow_status_sensor materializes this asset when telemetry shows activity. Matches the pattern of external_precisely_job + external_huggingface_space. All 6 follow the shared auth pattern: account / user / (password | PAT | keypair) / warehouse / role via *_env_var fields. The 3 action components support all three Snowflake auth methods; the 2 sensors inherit the same. Manifest + lazy-loader regenerated; total registry entries: 803. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent fb22334 commit 6171dfb

38 files changed

Lines changed: 1667 additions & 0 deletions

File tree

‎assets/data-warehouse/snowflake_dynamic_table_refresh_asset/README.md‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
# Snowflake Dynamic Table — Refresh Asset
2+
3+
Materializes by calling `ALTER DYNAMIC TABLE <db>.<schema>.<name> REFRESH` on a named existing Snowflake dynamic table. Pulls scheduling state + last-refresh-status into materialization metadata.
4+
5+
Single-entity counterpart to the auto-discovered dynamic-table assets in `snowflake_workspace`.
6+
7+
## Example
8+
9+
```yaml
10+
type: dagster_community_components.SnowflakeDynamicTableRefreshAssetComponent
11+
attributes:
12+
asset_key: snowflake/dt/customer_summary
13+
dynamic_table_name: CUSTOMER_SUMMARY
14+
database: DAGSTER_DEMO
15+
schema: ANALYTICS
16+
account_env_var: SNOWFLAKE_ACCOUNT
17+
user_env_var: SNOWFLAKE_USER
18+
password_env_var: SNOWFLAKE_PASSWORD
19+
```
20+
21+
## Materialization metadata
22+
23+
| Key | Description |
24+
|---|---|
25+
| `dynamic_table_fqn` | `<db>.<schema>.<name>` |
26+
| `scheduling_state` | RUNNING / SUSPENDED / SUSPENDED_ON_ERROR |
27+
| `last_refresh_status` | SUCCESS / FAILED / CANCELED |
28+
| `target_lag` | Configured lag target |
29+
| `refresh_mode` | AUTO / FULL / INCREMENTAL |
30+
| `last_successful_refresh_time` | ISO timestamp |
31+
32+
## Auth options
33+
34+
Same as `snowflake_task_execute_asset` — password / PAT / keypair.
35+
36+
## Requirements
37+
38+
```
39+
snowflake-connector-python>=3.0.0
40+
```

‎assets/data-warehouse/snowflake_dynamic_table_refresh_asset/__init__.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
from .component import SnowflakeDynamicTableRefreshAssetComponent
2+
3+
__all__ = ["SnowflakeDynamicTableRefreshAssetComponent"]

‎assets/data-warehouse/snowflake_dynamic_table_refresh_asset/component.py‎

Lines changed: 156 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,156 @@
1+
"""Snowflake Dynamic Table — refresh asset.
2+
3+
Materializes by calling ``ALTER DYNAMIC TABLE <db>.<schema>.<name> REFRESH``
4+
on a named, existing Snowflake dynamic table. Pulls scheduling_state +
5+
last_refresh_status into materialization metadata.
6+
7+
Single-entity counterpart to the auto-discovered dynamic-table assets in
8+
``snowflake_workspace``.
9+
"""
10+
from typing import Any, Dict, List, Optional
11+
12+
import dagster as dg
13+
from dagster import (
14+
AssetExecutionContext,
15+
Component,
16+
ComponentLoadContext,
17+
Definitions,
18+
MaterializeResult,
19+
Model,
20+
Resolvable,
21+
asset,
22+
)
23+
from pydantic import Field
24+
25+
26+
class SnowflakeDynamicTableRefreshAssetComponent(Component, Model, Resolvable):
27+
"""Materialize by ALTER DYNAMIC TABLE ... REFRESH on a named dynamic table.
28+
29+
Example:
30+
```yaml
31+
type: dagster_community_components.SnowflakeDynamicTableRefreshAssetComponent
32+
attributes:
33+
asset_key: snowflake/dt/customer_summary
34+
dynamic_table_name: CUSTOMER_SUMMARY
35+
database: DAGSTER_DEMO
36+
schema: ANALYTICS
37+
account_env_var: SNOWFLAKE_ACCOUNT
38+
user_env_var: SNOWFLAKE_USER
39+
password_env_var: SNOWFLAKE_PASSWORD
40+
```
41+
"""
42+
43+
asset_key: str = Field(description="Dagster asset key (e.g. 'snowflake/dt/customer_summary').")
44+
dynamic_table_name: str = Field(description="Name of the dynamic table.")
45+
database: str = Field(description="Snowflake database holding the DT.")
46+
schema_name: str = Field(description="Snowflake schema holding the DT.", alias="schema")
47+
48+
account_env_var: str = Field(description="Env var with Snowflake account.")
49+
user_env_var: str = Field(description="Env var with Snowflake username.")
50+
password_env_var: Optional[str] = Field(default=None, description="Env var with Snowflake password.")
51+
pat_env_var: Optional[str] = Field(default=None, description="Env var with Snowflake PAT.")
52+
private_key_path_env_var: Optional[str] = Field(default=None, description="Env var with RSA private key path.")
53+
private_key_passphrase_env_var: Optional[str] = Field(default=None, description="Env var with private-key passphrase.")
54+
warehouse_env_var: Optional[str] = Field(default=None, description="Env var with Snowflake warehouse.")
55+
role_env_var: Optional[str] = Field(default=None, description="Env var with Snowflake role.")
56+
57+
group_name: Optional[str] = Field(default="snowflake", description="Dagster asset group name.")
58+
description: Optional[str] = Field(default=None, description="Asset description.")
59+
owners: Optional[List[str]] = Field(default=None, description="Asset owners.")
60+
asset_tags: Optional[Dict[str, str]] = Field(default=None, description="Catalog tags.")
61+
kinds: Optional[List[str]] = Field(default=None, description="Asset kinds (auto-includes 'snowflake').")
62+
deps: Optional[List[str]] = Field(default=None, description="Upstream Dagster asset keys.")
63+
64+
model_config = {"populate_by_name": True}
65+
66+
def _connection_kwargs(self) -> Dict[str, Any]:
67+
import os
68+
kwargs: Dict[str, Any] = {
69+
"account": os.environ[self.account_env_var],
70+
"user": os.environ[self.user_env_var],
71+
"database": self.database,
72+
"schema": self.schema_name,
73+
}
74+
if self.warehouse_env_var:
75+
kwargs["warehouse"] = os.environ.get(self.warehouse_env_var, "")
76+
if self.role_env_var:
77+
kwargs["role"] = os.environ.get(self.role_env_var, "")
78+
if self.pat_env_var and os.environ.get(self.pat_env_var):
79+
kwargs["password"] = os.environ[self.pat_env_var]
80+
elif self.private_key_path_env_var and os.environ.get(self.private_key_path_env_var):
81+
kwargs["private_key_file"] = os.environ[self.private_key_path_env_var]
82+
if self.private_key_passphrase_env_var:
83+
kwargs["private_key_file_pwd"] = os.environ.get(self.private_key_passphrase_env_var, "")
84+
kwargs["authenticator"] = "snowflake_jwt"
85+
elif self.password_env_var:
86+
kwargs["password"] = os.environ.get(self.password_env_var, "")
87+
return kwargs
88+
89+
def build_defs(self, context: ComponentLoadContext) -> Definitions:
90+
_self = self
91+
kinds = set(self.kinds) if self.kinds else set()
92+
kinds.add("snowflake")
93+
94+
@asset(
95+
key=dg.AssetKey(self.asset_key.split("/")),
96+
description=self.description or (
97+
f"Snowflake dynamic table: "
98+
f"{self.database}.{self.schema_name}.{self.dynamic_table_name}"
99+
),
100+
group_name=self.group_name,
101+
kinds=kinds,
102+
owners=self.owners,
103+
tags=self.asset_tags,
104+
deps=[dg.AssetKey.from_user_string(k) for k in (self.deps or [])],
105+
)
106+
def snowflake_dynamic_table_refresh_asset(context: AssetExecutionContext) -> MaterializeResult:
107+
try:
108+
import snowflake.connector
109+
except ImportError:
110+
raise ImportError(
111+
"snowflake_dynamic_table_refresh_asset requires "
112+
"'snowflake-connector-python'. Install with: "
113+
"pip install snowflake-connector-python"
114+
)
115+
116+
fqn = f"{_self.database}.{_self.schema_name}.{_self.dynamic_table_name}"
117+
conn = snowflake.connector.connect(**_self._connection_kwargs())
118+
cursor = conn.cursor()
119+
try:
120+
context.log.info(f"ALTER DYNAMIC TABLE {fqn} REFRESH")
121+
cursor.execute(f"ALTER DYNAMIC TABLE {fqn} REFRESH")
122+
123+
# Pull current state from INFORMATION_SCHEMA.DYNAMIC_TABLES.
124+
cursor.execute(f"""
125+
SELECT name, scheduling_state, last_refresh_status,
126+
target_lag, refresh_mode,
127+
last_successful_refresh_time
128+
FROM {_self.database}.INFORMATION_SCHEMA.DYNAMIC_TABLES
129+
WHERE schema_name = '{_self.schema_name}'
130+
AND name = '{_self.dynamic_table_name}'
131+
""")
132+
row = cursor.fetchone()
133+
134+
metadata: Dict[str, Any] = {
135+
"dynamic_table_fqn": fqn,
136+
"dynamic_table_name": _self.dynamic_table_name,
137+
"database": _self.database,
138+
"schema": _self.schema_name,
139+
}
140+
if row:
141+
columns = [c[0] for c in cursor.description]
142+
rd = dict(zip(columns, row))
143+
metadata.update({
144+
"scheduling_state": rd.get("SCHEDULING_STATE"),
145+
"last_refresh_status": rd.get("LAST_REFRESH_STATUS"),
146+
"target_lag": rd.get("TARGET_LAG"),
147+
"refresh_mode": rd.get("REFRESH_MODE"),
148+
"last_successful_refresh_time": str(rd.get("LAST_SUCCESSFUL_REFRESH_TIME"))
149+
if rd.get("LAST_SUCCESSFUL_REFRESH_TIME") else None,
150+
})
151+
return MaterializeResult(metadata=metadata)
152+
finally:
153+
cursor.close()
154+
conn.close()
155+
156+
return Definitions(assets=[snowflake_dynamic_table_refresh_asset])

‎assets/data-warehouse/snowflake_dynamic_table_refresh_asset/example.yaml‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
type: dagster_community_components.SnowflakeDynamicTableRefreshAssetComponent
2+
attributes:
3+
asset_key: snowflake/dt/customer_summary
4+
dynamic_table_name: CUSTOMER_SUMMARY
5+
database: DAGSTER_DEMO
6+
schema: ANALYTICS
7+
account_env_var: SNOWFLAKE_ACCOUNT
8+
user_env_var: SNOWFLAKE_USER
9+
password_env_var: SNOWFLAKE_PASSWORD
10+
warehouse_env_var: SNOWFLAKE_WAREHOUSE
11+
role_env_var: SNOWFLAKE_ROLE
12+
group_name: snowflake

‎assets/data-warehouse/snowflake_dynamic_table_refresh_asset/requirements.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
snowflake-connector-python>=3.0.0

‎assets/data-warehouse/snowflake_dynamic_table_refresh_asset/schema.json‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
{
2+
"component_type": "dagster_community_components.SnowflakeDynamicTableRefreshAssetComponent",
3+
"description": "Materialize by ALTER DYNAMIC TABLE ... REFRESH on a named Snowflake dynamic table. Single-entity counterpart to snowflake_workspace's auto-discovered DTs.",
4+
"category": "data-warehouse",
5+
"icon": "si:snowflake",
6+
"x-dagster-io": {
7+
"outputs": {
8+
"type": "MaterializeResult",
9+
"description": "Snowflake DT refresh result + last status"
10+
}
11+
},
12+
"attributes": {
13+
"asset_key": { "type": "string", "label": "Asset Key", "required": true },
14+
"dynamic_table_name": { "type": "string", "label": "Dynamic Table Name", "required": true },
15+
"database": { "type": "string", "label": "Database", "required": true },
16+
"schema": { "type": "string", "label": "Schema", "required": true },
17+
"account_env_var": { "type": "string", "label": "Account Env Var", "required": true, "ui:widget": "env_var" },
18+
"user_env_var": { "type": "string", "label": "User Env Var", "required": true, "ui:widget": "env_var" },
19+
"password_env_var": { "type": "string", "label": "Password Env Var", "required": false, "ui:widget": "env_var" },
20+
"pat_env_var": { "type": "string", "label": "PAT Env Var", "required": false, "ui:widget": "env_var" },
21+
"private_key_path_env_var": { "type": "string", "label": "Private Key Path Env Var", "required": false, "ui:widget": "env_var" },
22+
"private_key_passphrase_env_var": { "type": "string", "label": "Private Key Passphrase Env Var", "required": false, "ui:widget": "env_var" },
23+
"warehouse_env_var": { "type": "string", "label": "Warehouse Env Var", "required": false, "ui:widget": "env_var" },
24+
"role_env_var": { "type": "string", "label": "Role Env Var", "required": false, "ui:widget": "env_var" },
25+
"group_name": { "type": "string", "label": "Group Name", "required": false, "default": "snowflake" },
26+
"description": { "type": "string", "label": "Description", "required": false },
27+
"owners": { "type": "array", "items": { "type": "string" }, "label": "Owners", "required": false, "ui:widget": "list" },
28+
"asset_tags": { "type": "object", "label": "Asset Tags", "required": false, "ui:widget": "key-value" },
29+
"kinds": { "type": "array", "items": { "type": "string" }, "label": "Kinds", "required": false, "ui:widget": "list" },
30+
"deps": { "type": "array", "items": { "type": "string" }, "label": "Deps", "required": false, "ui:widget": "list" }
31+
},
32+
"tags": ["data-warehouse", "snowflake", "dynamic-table", "refresh"],
33+
"name": "Snowflake Dynamic Table Refresh Asset"
34+
}

‎assets/data-warehouse/snowflake_stored_procedure_call_asset/README.md‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
# Snowflake Stored Procedure — Call Asset
2+
3+
Materializes by calling `CALL <db>.<schema>.<proc>(args)` on a named existing Snowflake stored procedure. Single-entity counterpart to the auto-discovered stored-procedure assets in `snowflake_workspace`.
4+
5+
## Example
6+
7+
```yaml
8+
type: dagster_community_components.SnowflakeStoredProcedureCallAssetComponent
9+
attributes:
10+
asset_key: snowflake/procs/transform_customers
11+
procedure_name: TRANSFORM_CUSTOMERS
12+
database: DAGSTER_DEMO
13+
schema: ANALYTICS
14+
arguments: []
15+
account_env_var: SNOWFLAKE_ACCOUNT
16+
user_env_var: SNOWFLAKE_USER
17+
password_env_var: SNOWFLAKE_PASSWORD
18+
```
19+
20+
### With arguments
21+
22+
Pass positional arguments to the proc. Strings are auto-quoted; numbers / booleans are inlined:
23+
24+
```yaml
25+
attributes:
26+
procedure_name: REFRESH_BY_PARTITION
27+
arguments:
28+
- "2026-05-22" # quoted as a string literal
29+
- 100 # inlined as a number
30+
- true # inlined as TRUE
31+
```
32+
33+
## Auth options
34+
35+
Same as `snowflake_task_execute_asset` — password / PAT / keypair via `*_env_var` fields.
36+
37+
## Requirements
38+
39+
```
40+
snowflake-connector-python>=3.0.0
41+
```
42+
43+
## See also
44+
45+
- [`snowflake_workspace`](https://dagster-component-ui.vercel.app/c/snowflake_workspace) — multi-entity auto-discovery
46+
- [`snowflake_task_execute_asset`](https://dagster-community-components-cli/c/snowflake_task_execute_asset) — sibling for tasks

‎assets/data-warehouse/snowflake_stored_procedure_call_asset/__init__.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
from .component import SnowflakeStoredProcedureCallAssetComponent
2+
3+
__all__ = ["SnowflakeStoredProcedureCallAssetComponent"]

0 commit comments

Comments
 (0)