---
name: add-dlt-pii-hashing
description: Add salted SHA-256 PII column hashing to a dlt connector — pseudonymize named columns at extraction time, driven by config. Use when the user wants to hash or pseudonymize PII columns, anonymize emails/phone numbers/names, add hash_columns_config, set up a transform.hash_salt secret, or apply GDPR-style column masking to a connector.
argument-hint: "[connector-name]"
disable-model-invocation: true
---

# Add PII Column Hashing to a dlt Connector

Add salted SHA-256 PII column hashing to the connector `$ARGUMENTS`, following the pattern already used by the `shopify` and `recharge` connectors.

## Context

This wires up **pseudonymization at extraction time**: named columns are replaced with `sha256(salt + canonical_value)` via a dlt `add_map` transform that runs *before* dlt normalizes field names. Which columns get hashed is driven by config (`hash_columns_config`), so operators can change it without touching code.

Connectors live in `connectors/$ARGUMENTS/` within the dlt-connectors monorepo (for development) or in a client Airflow repo (synced via `sync-connectors.sh`). The hashing is deterministic (same input → same hash with a fixed salt), so merge keys and joins on hashed columns keep working across runs.

### What gets created / changed

1. **Transformation module** — the `hash_columns` / `require_hash_salt` functions.
2. **Source wiring** — a `hash_columns_config` param, fail-fast salt check, and per-resource `add_map`.
3. **Env-var override** — `HASH_COLUMNS_CONFIG` for Airflow/Docker where editing `config.toml` is awkward.
4. **Secrets** — the `transform.hash_salt` secret in `secrets_example.toml`.
5. **Config** — a `[sources.<name>.hash_columns_config]` baseline in `config.toml`.

## Step 1 — Transformation module

Decide the file path by the connector's layout:

- **Package layout** (a `connectors/$ARGUMENTS/$ARGUMENTS/` source module exists, like `shopify`): create `connectors/$ARGUMENTS/$ARGUMENTS/transformations.py`. Import with `from $ARGUMENTS.transformations import hash_columns, require_hash_salt`.
- **Flat layout** (the pipeline is the only module, like `recharge`): create `connectors/$ARGUMENTS/${ARGUMENTS}_transformations.py`. Import with `from ${ARGUMENTS}_transformations import hash_columns, require_hash_salt`.

Write this module verbatim (adjust the field-name hint in the two warnings to match the API: snake_case for JSON APIs like Recharge, camelCase for GraphQL like Shopify):

```python
"""Row-level transformations for the $ARGUMENTS connector."""

from __future__ import annotations

import hashlib
import json
import logging
from typing import Any

import dlt

logger = logging.getLogger(__name__)

_hash_salt: str | None = None
_HASHABLE_TYPES = (str, bytes, dict, list)
_warned_columns: set[str] = set()


def _get_hash_salt() -> str:
    global _hash_salt
    if _hash_salt is None:
        _hash_salt = dlt.secrets["transform.hash_salt"]
    return _hash_salt


def require_hash_salt() -> str:
    """Resolve the hash salt eagerly so a misconfiguration fails at startup,
    not on the first hashed record mid-load. Raises ValueError if unset.

    Also resets the one-time-warning state so each pipeline run (each source
    instantiation) starts fresh, even in a long-lived process (e.g. an Airflow
    worker) that re-runs the same connector in the same interpreter.
    """
    _warned_columns.clear()
    try:
        return _get_hash_salt()
    except Exception as exc:
        raise ValueError(
            "hash_columns_config is set but the 'transform.hash_salt' secret is "
            "missing. Add it to .dlt/secrets.toml (see secrets_example.toml), or "
            "remove hash_columns_config to disable hashing."
        ) from exc


def _canonical(value: Any) -> str:
    if isinstance(value, (dict, list)):
        return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
    return str(value)


def hash_columns(columns: list[str], record: dict) -> dict:
    """Salted SHA-256 the given columns of a record (PII pseudonymization).

    Column names are the raw API field names. This map runs during extraction
    before dlt normalizes names, so config column names must match the API
    field names one-to-one. Null/empty values are left untouched. Non-string
    scalars (e.g. ints, decimals) are skipped so the column's inferred type is
    not changed. A one-time warning is logged per skipped or missing column.
    """
    salt = _get_hash_salt()
    record = dict(record)
    for column in columns:
        if column not in record:
            if column not in _warned_columns:
                logger.warning(
                    "hash_columns: column %r not present in records, so nothing is "
                    "hashed for it. Use the raw API field name.",
                    column,
                )
                _warned_columns.add(column)
            continue
        value = record.get(column)
        if value is None or value == "" or value == [] or value == {}:
            continue
        if not isinstance(value, _HASHABLE_TYPES):
            if column not in _warned_columns:
                logger.warning(
                    "hash_columns: skipping non-string scalar column %r (type %s) "
                    "to avoid changing the column's inferred type",
                    column,
                    type(value).__name__,
                )
                _warned_columns.add(column)
            continue
        record[column] = hashlib.sha256(
            f"{salt}{_canonical(value)}".encode()
        ).hexdigest()
    return record
```

Key design points to preserve:

- **Salt + SHA-256 hex digest.** The salt comes from the `transform.hash_salt` secret. Same input → same hash (stable across runs, so merge keys and joins still work), but not reversible without the salt.
- **Canonicalization.** Dicts/lists are JSON-dumped with `sort_keys=True` and compact separators so the hash is deterministic regardless of key order. Scalars are stringified.
- **Type preservation.** Numeric/boolean scalars are skipped so hashing never flips a column's inferred type (which would break `schema_contract={"data_type": "freeze"}` or create `__v_text` variant columns). Hash only string/bytes/dict/list.
- **Null/empty passthrough.** `None`, `""`, `[]`, `{}` are left as-is so absent values don't all collapse to one hash.
- **One-time warnings.** Missing or skipped columns warn once (tracked in `_warned_columns`) instead of flooding logs per record. Missing-column warnings are the usual symptom of using the *destination* (normalized) name instead of the raw API field name. `require_hash_salt()` clears this set at the start of each run, so the once-only suppression resets per pipeline invocation rather than persisting across runs in a long-lived process.
- **Shared salt, resolved once.** `_get_hash_salt()` caches `transform.hash_salt` at module level, so every source in the process shares one salt. This matches the Gemma convention of a single salt per destination (shared across stores/accounts — see Step 5). Do not rely on per-store salts within the same process: the first resolved salt wins.

## Step 2 — Wire it into the source

In `connectors/$ARGUMENTS/${ARGUMENTS}_pipeline.py`:

Add the import (use the path from Step 1):
```python
from functools import partial
from $ARGUMENTS.transformations import hash_columns, require_hash_salt  # or ${ARGUMENTS}_transformations
```

Add the parameter to the `@dlt.source` function and a fail-fast salt check:
```python
@dlt.source(name=DLT_SOURCE_NAME)
def ${ARGUMENTS}_source(
    # ...existing params...
    hash_columns_config: dict[str, list[str]] | None = None,
):
    hash_cfg = hash_columns_config or {}
    if any(cols for cols in hash_cfg.values()):
        # Fail fast before any network call: resolve the salt now rather than
        # dying on the first hashed record mid-load.
        require_hash_salt()
```

Apply the map at **resource scope** — one `add_map` per resource, keyed by `resource.name`. Two correct placements (pick the one that fits the connector):

**(a) Inside the source, as each resource is built** — the `shopify` pattern, preferred when you control the source body. Add a small decorator and apply it to every resource you yield:
```python
    def _decorate(resource):
        cols = hash_cfg.get(resource.name)
        if cols:
            resource.add_map(partial(hash_columns, cols))
        return resource

    # then yield the decorated resource for each one, e.g.:
    #   yield _decorate(customers())
```

**(b) After instantiation**, iterating the source's resources (e.g. in `run_pipeline`, before `pipeline.run`):
```python
    source = ${ARGUMENTS}_source(**kwargs)
    for resource in source.resources.values():
        cols = hash_cfg.get(resource.name)
        if cols:
            resource.add_map(partial(hash_columns, cols))
```
Applying it at the wrong scope (e.g. once on the source instead of per resource) silently hashes nothing.

`hash_columns_config` is auto-injected by dlt from `config.toml` (matched by `DLT_SOURCE_NAME`), so the source picks up the baseline with no extra plumbing.

## Step 3 — Env-var override (run-time)

Near the top of the pipeline module:
```python
# Run-time PII hashing override. When set, a JSON map of {resource: [columns]}
# (e.g. '{"customers": ["email", "phone"]}') that REPLACES the config.toml
# baseline ([sources.<name>.hash_columns_config]) for this run. Unset → fall
# back to config.toml. For Airflow/Docker where editing config.toml is awkward.
HASH_COLUMNS_CONFIG = os.getenv("HASH_COLUMNS_CONFIG")
```

In the run/`__main__` block, pass it explicitly so it wins over dlt's config injection (replace, not merge):
```python
    kwargs: dict = {}  # ...existing kwargs...
    if HASH_COLUMNS_CONFIG:
        try:
            kwargs["hash_columns_config"] = json.loads(HASH_COLUMNS_CONFIG)
        except json.JSONDecodeError as exc:
            raise ValueError(
                "HASH_COLUMNS_CONFIG must be valid JSON, e.g. "
                '\'{"customers": ["email", "phone"]}\''
            ) from exc

    load_info = pipeline.run(${ARGUMENTS}_source(**kwargs))
```

## Step 4 — Secrets

Add to `connectors/$ARGUMENTS/.dlt/secrets_example.toml`:
```toml
[transform]
hash_salt = "your-random-string-here"
```

Generate a strong random salt for the real `secrets.toml` (e.g. `openssl rand -hex 32`) and **keep it stable** — changing the salt changes every hash, so existing rows won't match new ones. Never read or commit the real `secrets.toml`.

## Step 5 — Config baseline

Add the columns to hash per resource in `connectors/$ARGUMENTS/.dlt/config.toml`, keyed by the source name (`DLT_SOURCE_NAME`). Use the **raw API field names** (the names as they arrive from the API, before dlt normalization), not the snake_case destination column names:
```toml
[sources.$ARGUMENTS.hash_columns_config]
customers = ["email", "firstName", "lastName", "phone"]
orders    = ["email", "phone", "billingAddress", "shippingAddress"]
```

For multi-store/multi-account setups, repeat the block per source name (`[sources.shopify_de.hash_columns_config]`, `[sources.shopify_ch.hash_columns_config]`, ...).

## Step 6 — Verify

- Run the pipeline and confirm hashed columns contain 64-char hex strings, not plaintext.
- Confirm a missing-column warning appears exactly once if you misname a column (good signal you used the destination name instead of the raw API name).
- Confirm the run fails fast with the `require_hash_salt` ValueError if `hash_columns_config` is set but the salt is absent.

```bash
cd connectors/$ARGUMENTS
uv run --env-file=.env python ${ARGUMENTS}_pipeline.py
```

Before committing, run `just fmt && just lint`.