Gerenciamento de segredos e configuração do app Streamlit¶

Gerenciamento da conexão com o Snowflake¶

Para gerenciar sua conexão com o Snowflake, você pode usar st.connection("snowflake"). Isso permite que você se conecte ao Snowflake do seu ambiente de desenvolvimento local e do seu app implantado.

import streamlit as st

conn = st.connection("snowflake")
session = conn.session()

session.sql("SELECT 1").collect()

Nos tempos de execução de warehouse, você também pode usar a função get_active_session() do Snowpark para obter a sessão ativa.

import streamlit as st
from snowflake.snowpark.context import get_active_session

# ONLY IN WAREHOUSE RUNTIMES
session = get_active_session()
session.sql("SELECT 1").collect()

Segredos em tempos de execução de contêiner¶

Os tempos de execução de contêiner não têm acesso ao módulo _snowflake porque eles são executados fora do ambiente do procedimento armazenado. Para acessar segredos em um tempo de execução de contêiner, você deve criar funções SQL que usem o módulo _snowflake e, em seguida, chamar essas funções do seu app Streamlit. Para chamadas da Cortex API, você deve usar requests.

Se você atualizar um app Streamlit mais antigo para um tempo de execução de contêiner, as funções _snowflake a seguir deverão ser substituídas por procedimentos armazenados. Esse procedimento está descrito na próxima seção.

• get_generic_secret_string

• get_oauth_access_token

• get_username_password

• get_cloud_provider_token

• get_secret_type

Além disso, a função _snowflake a seguir deve ser substituída por uma chamada de API manual, autenticada com um token de sessão. Esse procedimento está descrito em uma seção posterior, Chamada de um Cortex Agent em tempo de execução de contêiner.

• send_snow_api_request

# .streamlit/secrets.toml
# ONLY USE FOR NON-SECRET CONFIGURATION VALUES
app_name = "My Streamlit App"
api_endpoint = "https://api.example.com"
max_results = 100

import streamlit as st
import os

app_name = st.secrets["app_name"]
API_ENDPOINT = os.getenv("API_ENDPOINT")

import requests
import json
import os

SNOWFLAKE_HOST = os.getenv("SNOWFLAKE_HOST")
SNOWFLAKE_ACCOUNT = os.getenv("SNOWFLAKE_ACCOUNT")
ANALYST_ENDPOINT = "/api/v2/cortex/analyst/message"
URL = "https://" + SNOWFLAKE_HOST + ANALYST_ENDPOINT

def get_token() -> str:
    """Read the oauth token embedded into SPCS container"""
    return open("/snowflake/session/token", "r").read()

def send_request(semantic_model_file, prompt):
    """Sends the prompt using the semantic model file """
    headers = {
        "Content-Type": "application/json",
        "accept": "application/json",
        "Authorization": f"Bearer {get_token()}",
        "X-Snowflake-Authorization-Token-Type": "OAUTH"
    }
    request_body = {
        "messages": [
            {
                "role": "user",
                "content": [{"type": "text", "text": prompt}],
            }
        ],
        "semantic_model_file": semantic_model_file,
    }
    return requests.post(URL, headers=headers, data=json.dumps(request_body))

Segredos em tempos de execução de warehouse¶

Em tempos de execução de warehouse, você pode usar o módulo _snowflake para acessar segredos diretamente no código do app Streamlit. Os tempos de execução de warehouse herdam o acesso ao módulo _snowflake dos procedimentos armazenados, o que permite recuperar segredos referenciados no objeto Streamlit.

Para usar segredos em um tempo de execução de warehouse:

1. Crie um objeto de segredo no Snowflake. Para obter mais informações, consulte CREATE SECRET.

   CREATE OR REPLACE SECRET my_secret
     TYPE = GENERIC_STRING
     SECRET_STRING = 'my_secret_value';
   

   Copy

2. Crie uma integração de acesso externo e atribua o segredo a ela.

   CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION my_eai
     ALLOWED_AUTHENTICATION_SECRETS = (my_secret)
     ENABLED = TRUE;
   

   Copy

3. Faça referência ao segredo em seu objeto Streamlit usando o parâmetro SECRETS:

   ALTER STREAMLIT my_warehouse_app
     SET EXTERNAL_ACCESS_INTEGRATIONS = (my_eai)
     SECRETS = ('my_secret_key' = my_secret);
   

   Copy

   Você deve atribuir a integração de acesso externo e o segredo ao objeto Streamlit. Não é possível atribuir apenas o segredo a um objeto Streamlit.

4. No código do app Streamlit, importe o módulo _snowflake e recupere o segredo:

   import streamlit as st
   import _snowflake
   
   # Retrieve an API key from a generic string secret
   my_secret = _snowflake.get_generic_secret_string('my_secret_key')
   

   Copy

Para obter mais informações sobre como acessar segredos com o módulo _snowflake, consulte API de Python para acesso secreto.

Configuração do Streamlit¶

Os apps Streamlit podem incluir um arquivo de configuração (.streamlit/config.toml). Esse arquivo permite que você personalize vários aspectos do seu app, como tema, layout e comportamento. O arquivo de configuração é escrito no formato TOML. Para obter mais informações sobre as opções de configuração disponíveis, consulte a documentação do Streamlit em config.toml.

O suporte para opções de configuração varia de acordo com o ambiente de tempo de execução. Os tempos de execução de contêiner geralmente oferecem suporte mais amplo para opções de configuração do que os tempos de execução de warehouse, em especial para serviços estáticos. A tabela a seguir mostra quais seções de configuração são compatíveis com os tempos de execução de warehouse e de contêiner:

Para obter informações sobre como usar a seção [snowflake.sleep] para configurar temporizadores de suspensão em tempos de execução de warehouse, consulte Temporizador de sono personalizado para um aplicativo Streamlit.

A estrutura de diretório a seguir mostra um exemplo de um app Streamlit com um arquivo de configuração:

source_directory/
├── .streamlit/
│   └── config.toml
├── pyproject.toml
├── streamlit_app.py
└── uv.lock