# Snowflake & Redis Reference

## `<svc>/connectors/snowflake.py`

```python
import hashlib
from typing import Any

from ddtrace import tracer
from snowflake_connector import snowflake_conn

from <svc> import config

# Default connection args passed to every query
DEFAULT_CONFIG = {
    "account":   config.SNOWFLAKE_ACCOUNT,
    "user":      config.SNOWFLAKE_USER,
    "database":  config.SNOWFLAKE_DATABASE,
    "warehouse": config.SNOWFLAKE_WAREHOUSE,
    "role":      config.SNOWFLAKE_ROLE,
    "private_key": config.SNOWFLAKE_PRIVATE_KEY,
    "client_session_keep_alive": True,
}

SnowflakeParameterType = dict[str, Any] | list[Any] | None


@tracer.wrap(name="snowflake_fetchall")
def fetchall(query: str, params: SnowflakeParameterType = None) -> list[dict]:
    """Execute a query and return results as a list of dicts (cached via Redis if available)."""
    return snowflake_conn.fetchall(query, params, **DEFAULT_CONFIG)


def fetchall_nocache(query: str, params: SnowflakeParameterType = None) -> list[dict]:
    """Execute a query bypassing any cache layer (use for write-after-read consistency)."""
    return snowflake_conn.fetchall(query, params, **DEFAULT_CONFIG)
```

**Snowflake rules:**
- No ORM for Snowflake — raw SQL only. SQLAlchemy is only used as a connection pool.
- Pool size must equal uWSGI `--cheaper-initial` worker count (default 15 each).
- `pool_recycle` is `4 * 55 * 60` seconds (~4h 55min) — Snowflake's idle timeout is ~4h; recycle before that.
- `client_session_keep_alive=True` prevents session expiry on long queries.
- `SNOWFLAKE_HOME` env var prevents connector from writing to `/root` (which the `uwsgi` user cannot access).

## SQL Files

Store `.sql` files in `<svc>/queries/sql/` or alongside the logic module. Load them with Python's `Path(__file__).parent / "sql" / "my_query.sql"`:

```python
from pathlib import Path

_SQL_DIR = Path(__file__).parent / "sql"


def _load_sql(filename: str) -> str:
    return (_SQL_DIR / filename).read_text()


def get_resource(resource_id: str, permissions_filter: dict) -> list[dict]:
    sql = _load_sql("get_resource.sql")
    return snowflake.fetchall(sql, {"resource_id": resource_id, **permissions_filter})
```

## Result Mapping

```python
# <svc>/utils/db/mapper.py
def map_db_result_with_column_names(
    raw_db_result: list[tuple], column_names: list[str]
) -> list[dict]:
    return [{c: v for c, v in zip(column_names, row)} for row in raw_db_result]
```

## `<svc>/connectors/redis.py`

```python
import redis

from <svc> import config

if not config.REDIS_HOST:
    # fakeredis is a runtime dependency — no test mock needed for Redis
    import fakeredis
    client: redis.Redis = fakeredis.FakeStrictRedis()
else:
    client = redis.StrictRedis(
        host=config.REDIS_HOST,
        port=config.REDIS_PORT,
        db=0,
    )
```

**Redis rules:**
- `fakeredis` must be a **production** dependency (not dev-only) — it is used when `REDIS_HOST` is not set (local dev, test environments). If it were dev-only the deploy image would fail.
- Always `db=0` unless there is a specific reason to use a different database.
- Never open a new connection per request — the module-level `client` is shared.

## `<svc>/utils/cache.py` — Caching Decorator

```python
import gzip
import hashlib
import json
import logging
from collections.abc import Callable
from datetime import date, datetime
from decimal import Decimal
from functools import wraps
from typing import Any

from <svc>.connectors import redis as redis_connector

logger = logging.getLogger(__name__)

ONE_MINUTE = 60
FIVE_MINUTES = 5 * ONE_MINUTE
ONE_HOUR = 60 * ONE_MINUTE
ONE_DAY = 24 * ONE_HOUR
ONE_WEEK = 7 * ONE_DAY


def _serialize(obj: Any) -> str:
    def default(o: Any) -> Any:
        if isinstance(o, (date, datetime)):
            return o.isoformat()
        if isinstance(o, Decimal):
            return float(o)
        if hasattr(o, "__dict__"):
            return o.__dict__
        raise TypeError(f"Object of type {type(o)} is not JSON serializable")

    return json.dumps(obj, default=default)


def _make_key(func: Callable, args: tuple, kwargs: dict) -> str:
    raw = f"{func.__qualname__}:{repr(args)}:{repr(sorted(kwargs.items()))}"
    return hashlib.sha256(raw.encode()).hexdigest()


def cache_in_redis(ttl: int = ONE_HOUR, key: str | None = None):
    """Decorator that caches the return value of a function in Redis.

    Uses gzip-compressed JSON. Fails silently on Redis errors (logs warning).
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = key or _make_key(func, args, kwargs)
            try:
                cached = redis_connector.client.get(cache_key)
                if cached is not None:
                    return json.loads(gzip.decompress(cached))
            except Exception as e:
                logger.warning("Redis cache read failed: %s", e)

            result = func(*args, **kwargs)

            try:
                compressed = gzip.compress(json.dumps(result).encode())
                redis_connector.client.setex(cache_key, ttl, compressed)
            except Exception as e:
                logger.warning("Redis cache write failed: %s", e)

            return result
        return wrapper
    return decorator
```

**Usage:**
```python
from <svc>.utils.cache import cache_in_redis, ONE_HOUR

@cache_in_redis(ttl=ONE_HOUR)
def get_expensive_data(resource_id: str, date: str) -> list[dict]:
    ...
```

## Logic Layer Pattern

```python
# <svc>/logic/<domain>.py
from <svc>.connectors import snowflake
from <svc>.utils.cache import cache_in_redis, ONE_HOUR
from <svc>.utils.db.mapper import map_db_result_with_column_names


_COLUMN_NAMES = ["id", "name", "created_at"]

_GET_RESOURCE_SQL = """
    SELECT
        r.id,
        r.name,
        r.created_at
    FROM my_schema.resources r
    WHERE r.id = %(resource_id)s
    AND r.is_active = TRUE
"""


@cache_in_redis(ttl=ONE_HOUR)
def get_resource(resource_id: str) -> dict | None:
    results = snowflake.fetchall(_GET_RESOURCE_SQL, {"resource_id": resource_id})
    if not results:
        return None
    return map_db_result_with_column_names(results, _COLUMN_NAMES)[0]


def get_resources(limit: int = 100, offset: int = 0) -> list[dict]:
    sql = f"{_BASE_SQL} LIMIT %(limit)s OFFSET %(offset)s"
    results = snowflake.fetchall(sql, {"limit": limit, "offset": offset})
    return map_db_result_with_column_names(results, _COLUMN_NAMES)
```

**Logic layer rules:**
- Logic functions receive plain Python values (strings, ints, dicts) — never Flask's `request` object.
- Return plain dicts/lists — not marshmallow-serialized output (that happens in the schema layer).
- Cache at the logic layer, not the handler layer.
- SQL strings defined as module-level constants (not inline in function bodies) — makes them easy to find and test.
- For large/complex queries, move SQL to `.sql` files; for simple queries, module-level strings are fine.