Snowflake Platform Skill

pip install snowflake-cli
snow --version  # Should show 3.14.0+

# Interactive setup
snow connection add

# Or create ~/.snowflake/config.toml manually

[connections.default]
account = "orgname-accountname"
user = "USERNAME"
authenticator = "SNOWFLAKE_JWT"
private_key_path = "~/.snowflake/rsa_key.p8"

snow connection test -c default
snow sql -q "SELECT CURRENT_USER(), CURRENT_ACCOUNT()"

-- Simple prompt
SELECT SNOWFLAKE.CORTEX.COMPLETE(
    'llama3.1-70b',
    'Explain quantum computing in one sentence'
) AS response;

-- With conversation history
SELECT SNOWFLAKE.CORTEX.COMPLETE(
    'llama3.1-70b',
    [
        {'role': 'system', 'content': 'You are a helpful assistant'},
        {'role': 'user', 'content': 'What is Snowflake?'}
    ]
) AS response;

-- With options
SELECT SNOWFLAKE.CORTEX.COMPLETE(
    'mistral-large2',
    'Summarize this document',
    {'temperature': 0.3, 'max_tokens': 500}
) AS response;

-- Single text
SELECT SNOWFLAKE.CORTEX.SUMMARIZE(article_text) AS summary
FROM articles
LIMIT 10;

-- Aggregate across rows (no context window limit)
SELECT AI_SUMMARIZE_AGG(review_text) AS all_reviews_summary
FROM product_reviews
WHERE product_id = 123;

-- Translate to English (auto-detect source)
SELECT SNOWFLAKE.CORTEX.TRANSLATE(
    review_text,
    '',      -- Empty = auto-detect source language
    'en'     -- Target language
) AS translated
FROM international_reviews;

-- Explicit source language
SELECT AI_TRANSLATE(
    description,
    'es',    -- Source: Spanish
    'en'     -- Target: English
) AS translated
FROM spanish_products;

-- Filter with plain English
SELECT * FROM customer_feedback
WHERE AI_FILTER(
    feedback_text,
    'mentions shipping problems or delivery delays'
);

-- Combine with SQL predicates for maximum optimization
-- Query planner applies standard filters FIRST, then AI on smaller dataset
SELECT * FROM support_tickets
WHERE created_date > '2025-01-01'  -- Standard filter applied first
  AND AI_FILTER(description, 'customer is angry or frustrated');

-- Categorize support tickets
SELECT
    ticket_id,
    AI_CLASSIFY(
        description,
        ['billing', 'technical', 'shipping', 'other']
    ) AS category
FROM support_tickets;

-- This seemingly simple query can be expensive at scale
SELECT
    product_id,
    AI_COMPLETE('mistral-large2', 'Summarize: ' || review_text) as summary
FROM product_reviews  -- 1 billion rows
WHERE created_date > '2024-01-01';

-- Cost = (input tokens + output tokens) × row count × model rate
-- At scale, this adds up fast

SELECT CURRENT_ACCOUNT();  -- Returns: NZ90655

# Generate private key (PKCS#8 format required)
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out ~/.snowflake/rsa_key.p8 -nocrypt

# Generate public key
openssl rsa -in ~/.snowflake/rsa_key.p8 -pubout -out ~/.snowflake/rsa_key.pub

# Get fingerprint for JWT claims
openssl rsa -in ~/.snowflake/rsa_key.p8 -pubout -outform DER | \
  openssl dgst -sha256 -binary | openssl enc -base64

-- In Snowflake worksheet (requires ACCOUNTADMIN or SECURITYADMIN)
ALTER USER my_user SET RSA_PUBLIC_KEY='MIIBIjANBgkq...';

iss: ACCOUNT_LOCATOR.USERNAME.SHA256:fingerprint
sub: ACCOUNT_LOCATOR.USERNAME

iss: NZ90655.JEZWEB.SHA256:jpZO6LvU2SpKd8tE61OGfas5ZXpfHloiJd7XHLPDEEA=
sub: NZ90655.JEZWEB

# No special configuration needed inside SPCS containers
import snowflake.connector

# Auto-detects SPCS_TOKEN environment variable
conn = snowflake.connector.connect()

# Initialize project
snow init

# Execute SQL
snow sql -q "SELECT 1"
snow sql -f query.sql

# View logs
snow logs

# Development
snow app run              # Deploy and run locally
snow app deploy           # Upload to stage only
snow app teardown         # Remove app

# Versioning
snow app version create V1_0
snow app version list
snow app version drop V1_0

# Publishing
snow app publish --version V1_0 --patch 0

# Release Channels
snow app release-channel list
snow app release-channel add-version --channel ALPHA --version V1_0
snow app release-directive set default --version V1_0 --patch 0 --channel DEFAULT

snow streamlit deploy --replace
snow streamlit deploy --replace --open

snow stage list
snow stage copy @my_stage/file.txt ./local/

my_native_app/
├── snowflake.yml           # Project config
├── manifest.yml            # App manifest
├── setup_script.sql        # Installation script
├── app/
│   └── streamlit/
│       ├── environment.yml
│       └── streamlit_app.py
└── scripts/
    └── setup.sql

definition_version: 2

native_app:
  name: my_app
  package:
    name: my_app_pkg
    distribution: external    # For marketplace
  application:
    name: my_app
  source_stage: stage/dev
  artifacts:
    - src: manifest.yml
      dest: manifest.yml
    - src: setup_script.sql
      dest: setup_script.sql
    - src: app/streamlit/environment.yml
      dest: streamlit/environment.yml
    - src: app/streamlit/streamlit_app.py
      dest: streamlit/streamlit_app.py
  enable_release_channels: true  # For ALPHA/BETA channels

manifest_version: 1

artifacts:
  setup_script: setup_script.sql
  default_streamlit: streamlit/streamlit_app.py

# Note: Do NOT include privileges section - Native Apps can't declare privileges

-- 1. Create network rule (in a real database, NOT app package)
CREATE DATABASE IF NOT EXISTS MY_APP_UTILS;

CREATE OR REPLACE NETWORK RULE MY_APP_UTILS.PUBLIC.api_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('api.example.com:443');

-- 2. Create integration
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION my_app_integration
  ALLOWED_NETWORK_RULES = (MY_APP_UTILS.PUBLIC.api_rule)
  ENABLED = TRUE;

-- 3. Grant to app
GRANT USAGE ON INTEGRATION my_app_integration
  TO APPLICATION MY_APP;

-- 4. CRITICAL: Attach to Streamlit (must repeat after EVERY deploy!)
ALTER STREAMLIT MY_APP.config_schema.my_streamlit
  SET EXTERNAL_ACCESS_INTEGRATIONS = (my_app_integration);

-- 1. Create shared_data schema in app package
CREATE SCHEMA IF NOT EXISTS MY_APP_PKG.SHARED_DATA;

-- 2. Create views referencing external database
CREATE OR REPLACE VIEW MY_APP_PKG.SHARED_DATA.MY_VIEW AS
SELECT * FROM EXTERNAL_DB.SCHEMA.TABLE;

-- 3. Grant REFERENCE_USAGE (CRITICAL!)
GRANT REFERENCE_USAGE ON DATABASE EXTERNAL_DB
  TO SHARE IN APPLICATION PACKAGE MY_APP_PKG;

-- 4. Grant access to share
GRANT USAGE ON SCHEMA MY_APP_PKG.SHARED_DATA
  TO SHARE IN APPLICATION PACKAGE MY_APP_PKG;
GRANT SELECT ON ALL VIEWS IN SCHEMA MY_APP_PKG.SHARED_DATA
  TO SHARE IN APPLICATION PACKAGE MY_APP_PKG;

# 1. Deploy app
snow app run

# 2. Create version
snow app version create V1_0

# 3. Check security review status
snow app version list
# Wait for review_status = APPROVED

# 4. Set release directive
snow app release-directive set default --version V1_0 --patch 0 --channel DEFAULT

# 5. Create listing in Snowsight Provider Studio (UI only)

from snowflake.snowpark import Session

connection_params = {
    "account": "orgname-accountname",
    "user": "USERNAME",
    "password": "PASSWORD",  # Or use private_key_path
    "warehouse": "COMPUTE_WH",
    "database": "MY_DB",
    "schema": "PUBLIC"
}

session = Session.builder.configs(connection_params).create()

# Read table
df = session.table("MY_TABLE")

# Filter and select
result = df.filter(df["STATUS"] == "ACTIVE") \
           .select("ID", "NAME", "CREATED_AT") \
           .sort("CREATED_AT", ascending=False)

# Execute
result.show()

# Collect to Python
rows = result.collect()

# WRONG - dict() doesn't work on Snowpark Row
config = dict(result[0])

# CORRECT - Access columns explicitly
row = result[0]
config = {
    'COLUMN_A': row['COLUMN_A'],
    'COLUMN_B': row['COLUMN_B'],
}

# Before v4.2.0 - rowcount returns -1 for CTAS
cursor.execute("CREATE TABLE new_table AS SELECT * FROM source WHERE active = true")
print(cursor.rowcount)  # Returns -1 (not helpful!)

# After v4.2.0 - stats property shows actual row counts
cursor.execute("CREATE TABLE new_table AS SELECT * FROM source WHERE active = true")
print(cursor.stats)  # Returns {'rows_inserted': 1234, 'duplicates': 0, ...}

from snowflake.snowpark.functions import udf, sproc

# Register UDF
@udf(name="my_udf", replace=True)
def my_udf(x: int) -> int:
    return x * 2

# Register Stored Procedure
@sproc(name="my_sproc", replace=True)
def my_sproc(session: Session, table_name: str) -> str:
    df = session.table(table_name)
    count = df.count()
    return f"Row count: {count}"

https://{org-account}.snowflakecomputing.com/api/v2/statements

const headers = {
  'Authorization': `Bearer ${jwt}`,
  'Content-Type': 'application/json',
  'Accept': 'application/json',  // REQUIRED - "null" error if missing
  'User-Agent': 'MyApp/1.0',
};

// Submit returns statementHandle, not results
const submit = await fetch(url, { method: 'POST', headers, body });
const { statementHandle } = await submit.json();

// Poll until complete
while (true) {
  const status = await fetch(`${url}/${statementHandle}`, { headers });
  if (status.status === 200) break;  // Complete
  if (status.status === 202) {
    await sleep(2000);  // Still running
    continue;
  }
}

const response = await fetch(url, {
  signal: AbortSignal.timeout(30000),  // 30 seconds
  headers,
});

POST /api/v2/statements/{statementHandle}/cancel

# AVOID - creates new connection each iteration
for i in range(1000):
    conn = snowflake.connector.connect(...)
    cursor = conn.cursor()
    cursor.execute("SELECT 1")
    cursor.close()
    conn.close()

# BETTER - reuse connection
conn = snowflake.connector.connect(...)
cursor = conn.cursor()
for i in range(1000):
    cursor.execute("SELECT 1")
cursor.close()
conn.close()

import time
import snowflake.connector

def execute_with_retry(cursor, query, max_retries=3):
    for attempt in range(max_retries):
        try:
            return cursor.execute(query).fetchall()
        except snowflake.connector.errors.DatabaseError as e:
            if "throttled" in str(e).lower() and attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                time.sleep(wait_time)
            else:
                raise