---
name: roche-snowflake-dia
description: Connect a Python/FastAPI backend to the Roche Snowflake DIA account using WAM OAuth (password grant). Use when a new project needs to read Snowflake data at Roche (DIA account), needs the OAuth token flow, connection settings, or a query helper.
---

# Roche Snowflake DIA connection (WAM OAuth)

How to authenticate to the Roche **DIA** Snowflake account from a backend service.
Snowflake does **not** use a username/password directly — you exchange Roche SSO (WAM)
credentials for an OAuth access token via the `password` grant, then connect with
`authenticator="oauth"`.

## Requirements
- Network: **Roche network / VPN** (Snowflake + WAM are internal).
- `pip install snowflake-connector-python requests pydantic-settings`
- A DIA OAuth **client secret** (ask the DIA/data team) + your Roche SSO login.

## Environment variables (never hard-code secrets)
| Var | Purpose | Default |
| --- | --- | --- |
| `SSO_USERNAME` / `SSO_PASSWORD` | Roche WAM SSO credentials | — |
| `SNOWFLAKE_CLIENT_SECRET` | DIA OAuth client secret | — |
| `DIA_CLIENT_ID` | OAuth client id | `SNOWFLAKE` |
| `OAUTH_TOKEN_ENDPOINT` | WAM token endpoint | `https://wam.roche.com/as/token.oauth2` |
| `SNOWFLAKE_ACCOUNT` | DIA account | `roche_dia.eu-central-1` |
| `SNOWFLAKE_ROLE` | role | `PUBLIC` |
| `SNOWFLAKE_WAREHOUSE` | warehouse | `WH_DATA_CONSUMER_01` |
| `SNOWFLAKE_DATABASE` | database | `PPP_PROD` |

## OAuth token request (the key part)
```python
payload = {
    "client_id": DIA_CLIENT_ID,          # "SNOWFLAKE"
    "client_secret": DIA_CLIENT_SECRET,
    "grant_type": "password",
    "username": SSO_USERNAME,
    "password": SSO_PASSWORD,
    "scope": "session:role-any",
}
resp = requests.post(OAUTH_TOKEN_ENDPOINT, data=payload, timeout=30)
access_token = resp.json()["access_token"]   # expires_in ~ 3300s
```
Cache the token in-process until ~60s before `expires_in`; guard with a `threading.Lock`.

## Connect + query
```python
import snowflake.connector  # import lazily so the app boots without the connector/DB

conn = snowflake.connector.connect(
    account=SNOWFLAKE_ACCOUNT,
    authenticator="oauth",
    token=access_token,
    role=SNOWFLAKE_ROLE,
    warehouse=SNOWFLAKE_WAREHOUSE,
    database=SNOWFLAKE_DATABASE,
    schema=schema,                       # pass the schema per query
)
cur = conn.cursor()
cur.execute(sql, params or {})           # use %(name)s bind params, never f-string values
cols = [c[0] for c in cur.description]
rows = [dict(zip(cols, r)) for r in cur.fetchall()]
```

## Patterns that paid off
- **Lazy import** of `snowflake.connector` inside `connect()` → the API can boot (and
  serve `/health`) without the connector installed or the DB reachable. Data endpoints
  fail with a clear 502 instead of blocking startup.
- **Singleton client** (`get_client()`) holding the cached token across requests.
- **Config-driven schemas/tables** via `pydantic-settings` `Field(alias="ENV_NAME")` so
  dev/prod tables swap without code changes.
- Surface DB errors as HTTP **502** (`Data source error: {exc}`), not 500.
- Add a **TTL cache** in front of expensive queries (see the reference project's `TTLCache`).

## Reference implementation
`bar-dashboard/backend/app/snowflake_client.py` and `app/config.py`. The logic was ported
from the shared Roche `_get_oauth_token_dia` / `_get_connection` helpers used across the
`pat-e2e-report-automation` family of apps.