Snowflake Integrations with Python

Problem

You are building an interactive Python application which needs to act as the person visiting the application when accessing a private Snowflake resource. The content must authenticate to Snowflake using the viewer’s Snowflake credentials. For example, reading data from a Snowflake table that has row-level permissions and each content viewer has a different level of data access.

Solution

The examples below illustrate how to use the OAuth integrations feature of Posit Connect to delegate authorization to the data provider. These examples use the viewer’s OAuth access token to call the Snowflake API on behalf of the content viewer, allowing content to access private Snowflake resources as the content viewer.

When publishing the content to Connect make sure the following environment variables are set for the deployed content:

  • SNOWFLAKE_ACCOUNT
  • SNOWFLAKE_WAREHOUSE
requirements.txt
snowflake-connector-python==3.12.1
streamlit==1.37.0
posit-sdk>=0.4.1
app.py
# -*- coding: utf-8 -*-
# mypy: ignore-errors
import os

import pandas as pd
import snowflake.connector
import streamlit as st
from posit.connect.external.snowflake import PositAuthenticator

ACCOUNT = os.getenv("SNOWFLAKE_ACCOUNT")
WAREHOUSE = os.getenv("SNOWFLAKE_WAREHOUSE")

# USER is only required when running the example locally with external browser auth
USER = os.getenv("SNOWFLAKE_USER")

# https://docs.snowflake.com/en/user-guide/sample-data-using
DATABASE = os.getenv("SNOWFLAKE_DATABASE", "snowflake_sample_data")
SCHEMA = os.getenv("SNOWFLAKE_SCHEMA", "tpch_sf1")
TABLE = os.getenv("SNOWFLAKE_TABLE", "lineitem")

session_token = st.context.headers.get("Posit-Connect-User-Session-Token")
auth = PositAuthenticator(
    local_authenticator="EXTERNALBROWSER", user_session_token=session_token
)

con = snowflake.connector.connect(
    user=USER,
    account=ACCOUNT,
    warehouse=WAREHOUSE,
    database=DATABASE,
    schema=SCHEMA,
    authenticator=auth.authenticator,
    token=auth.token,
)

snowflake_user = con.cursor().execute("SELECT CURRENT_USER()").fetchone()
st.write(f"Hello, {snowflake_user[0]}!")

with st.spinner("Loading data from Snowflake..."):
    df = pd.read_sql_query(f"SELECT * FROM {TABLE} LIMIT 10", con)

st.dataframe(df)

Running the app locally

Terminal
export SNOWFLAKE_ACCOUNT = "<snowflake-account-identifier>"
export SNOWFLAKE_WAREHOUSE = "<snowflake-warehouse-name>"

# USER is only required when running the example locally with external browser auth
SNOWFLAKE_USER="<snowflake-username>" streamlit run app.py