Skip to content

session

pinky_streamlit.session

Snowflake session management for Streamlit in Snowflake apps.

In SiS production context: get_active_session() is called automatically. In local dev: configure a LocalSessionConfig pointing to a named connection in ~/.snowflake/connections.toml.

LocalSessionConfig dataclass

Named-connection config for local development fallback.

Parameters:

Name Type Description Default
connection_name str

Connection name defined in ~/.snowflake/connections.toml.

 required
role str | None

Role to activate after connecting (None = keep default).

 None
warehouse str | None

Warehouse to use (None = keep default).

 None
database str | None

Database to use (None = keep default).

 None
schema str | None

Schema to use (None = keep default).

 None
Source code in src/pinky_streamlit/core/session.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@dataclass
class LocalSessionConfig:
    """Named-connection config for local development fallback.

    Args:
        connection_name: Connection name defined in ~/.snowflake/connections.toml.
        role:            Role to activate after connecting (None = keep default).
        warehouse:       Warehouse to use (None = keep default).
        database:        Database to use (None = keep default).
        schema:          Schema to use (None = keep default).
    """

    connection_name: str
    role: str | None = field(default=None)
    warehouse: str | None = field(default=None)
    database: str | None = field(default=None)
    schema: str | None = field(default=None)

SessionInfo

Bases: NamedTuple

Return value of get_session() — supports both named access and positional unpacking.

Source code in src/pinky_streamlit/core/session.py 
24
25
26
27
28
29
class SessionInfo(NamedTuple):
    """Return value of get_session() — supports both named access and positional unpacking."""

    session: snowpark.Session
    login: str
    environment: str

get_session(local_config=None)

Return a Snowpark session, the current user login, and the environment tag.

In Streamlit in Snowflake: uses get_active_session() and st.user.login_name. In local dev: builds a session from local_config via a named connection.

Environment is derived from the database name suffix (e.g. MY_DB_QA → "QA", MY_DB_PROD → "PROD"). Returns "UNDEFINED" if the suffix cannot be resolved.

Parameters:

Name Type Description Default
local_config LocalSessionConfig | None

Optional fallback config for local development. Required when running outside Snowflake.

 None

Returns:

Type Description
SessionInfo

SessionInfo(session, login, environment) — supports positional unpacking.

Raises:

Type Description
RuntimeError

If outside Snowflake context and no local_config is provided.
Source code in src/pinky_streamlit/core/session.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def get_session(
    local_config: LocalSessionConfig | None = None,
) -> SessionInfo:
    """Return a Snowpark session, the current user login, and the environment tag.

    In Streamlit in Snowflake: uses get_active_session() and st.user.login_name.
    In local dev: builds a session from local_config via a named connection.

    Environment is derived from the database name suffix (e.g. MY_DB_QA → "QA",
    MY_DB_PROD → "PROD"). Returns "UNDEFINED" if the suffix cannot be resolved.

    Args:
        local_config: Optional fallback config for local development.
                      Required when running outside Snowflake.

    Returns:
        SessionInfo(session, login, environment) — supports positional unpacking.

    Raises:
        RuntimeError: If outside Snowflake context and no local_config is provided.
    """
    try:
        session = get_active_session()
        login = st.user.login_name
    except Exception:
        if local_config is None:
            raise RuntimeError(
                "Not running inside Snowflake and no LocalSessionConfig provided. "
                "Pass a LocalSessionConfig to get_session() for local development."
            ) from None
        try:
            session = snowpark.Session.builder.config(
                "connection_name", local_config.connection_name
            ).create()
            if local_config.role:
                session.use_role(local_config.role)
            if local_config.warehouse:
                session.use_warehouse(local_config.warehouse)
            if local_config.database:
                session.use_database(local_config.database)
            if local_config.schema:
                session.use_schema(local_config.schema)
        except Exception as e:
            logger.error("[get_session] Local connection failed: %s", e, exc_info=True)
            raise
        login = "LOCAL_USER"

    db_name = str(session.get_current_database()).replace('"', "")
    parts = db_name.split("_")
    environment = parts[-1] if len(parts) > 1 else "UNDEFINED"

    logger.info("[get_session] Session ready — user=%s, env=%s", login, environment)
    return SessionInfo(session=session, login=str(login), environment=environment)