Most teams discover PII leaks after a customer complaint or a compliance audit. By then, the data has already traversed your ETL pipeline, landed in application logs, and been indexed by three different observability tools.

Finding it after the fact is archaeology. What you actually need is a system that tracks PII the moment it enters your code, propagates that knowledge through every transformation, and blocks it from leaving through the wrong channels. This is not a solved problem, but it is a solvable one.

What “tracking PII” actually means

PII tracking is often conflated with data masking or access control. Those are related concerns, but they are downstream of the real problem. You cannot redact, encrypt, or restrict access to data you cannot locate.

Tracking means knowing, at every point in your pipeline, which fields contain sensitive data. Not guessing based on column names like email or ssn. Actually knowing, because the data carries a label that survives transformation, aggregation, and serialization.

This requires three things: a type system or metadata layer that can tag fields as sensitive, a propagation mechanism that carries those tags as data moves between functions and services, and enforcement points where you validate that tagged data is only written to approved destinations.

The architecture: tag, propagate, enforce

The cleanest implementations I have seen use a labeled data type at the language level. In Python, you can wrap values in a class that carries a sensitivity tag.

from dataclasses import dataclass
from enum import Enum, auto
from typing import Generic, TypeVar

T = TypeVar("T")

class Sensitivity(Enum):
    PLAIN = auto()
    PII = auto()
    PCI = auto()

@dataclass(frozen=True)
class Labeled(Generic[T]):
    value: T
    sensitivity: Sensitivity = Sensitivity.PLAIN

    def map(self, fn):
        return Labeled(fn(self.value), self.sensitivity)

When a user registers an account, you label the raw inputs at the boundary.

from labeled import Labeled, Sensitivity

def create_user(payload: dict) -> dict:
    email = Labeled(payload["email"], Sensitivity.PII)
    name = Labeled(payload["name"], Sensitivity.PII)
    age = Labeled(int(payload["age"]), Sensitivity.PLAIN)

    user = {
        "id": generate_uuid(),
        "email": email,
        "name": name,
        "age": age,
    }

    return user

The key insight is that email and name are now self-describing. Any function that receives a Labeled value can inspect its sensitivity without parsing the payload or guessing from the key name.

Propagating labels through transformations

Labeled data is only useful if the label survives your business logic. When you map, filter, or aggregate, you need rules for how sensitivity combines.

The simplest rule set is a join semilattice. If you merge two values, the result gets the more restrictive label.

class Sensitivity(Enum):
    PLAIN = 1
    PII = 2
    PCI = 3

    def join(self, other: "Sensitivity") -> "Sensitivity":
        return self if self.value >= other.value else other

When you serialize a record for a downstream service, you include the label in the metadata. The consumer can then decide whether to write to a raw database, a masked analytics warehouse, or an audit log.

def serialize(record: dict) -> dict:
    return {
        key: {
            "value": serialize_value(val.value),
            "sensitivity": val.sensitivity.name.lower(),
        }
        for key, val in record.items()
        if isinstance(val, Labeled)
    }

This is the part that trips people up. You cannot just tag at ingestion and hope for the best. Labels must be a first-class concern in your serialization layer. If your internal RPC format drops the metadata, your pipeline goes blind the moment data crosses a service boundary.

Enforcement: where the labels actually matter

Tracking without enforcement is expensive observability theater. You need choke points where labeled data is inspected before it is written to storage, sent over a network, or rendered in a UI.

A common pattern is a sink registry that maps destinations to maximum allowed sensitivity.

SINK_POLICIES = {
    "raw_postgres": {Sensitivity.PLAIN, Sensitivity.PII, Sensitivity.PCI},
    "analytics_clickhouse": {Sensitivity.PLAIN},
    "application_logs": {Sensitivity.PLAIN},
    "third_party_api": set(),
}

def write(sink_name: str, record: dict):
    allowed = SINK_POLICIES[sink_name]
    for key, val in record.items():
        if isinstance(val, Labeled) and val.sensitivity not in allowed:
            raise ValueError(
                f"Field {key} with sensitivity {val.sensitivity.name} "
                f"is not allowed in sink {sink_name}"
            )
    # proceed with write

If your analytics pipeline tries to ingest a record containing Sensitivity.PII into ClickHouse, the write fails at runtime with a clear error. This is not elegant, but it is explicit. Explicit is better than a GDPR violation.

For logging specifically, you can enforce this at the logger level. Python’s logging module supports filters that can inspect log records before they are emitted.

import logging

class PiiFilter(logging.Filter):
    def filter(self, record):
        msg = record.getMessage()
        # In practice, use a more sophisticated check or structured logging
        if any(label in msg for label in ["email", "ssn", "phone"]):
            record.msg = "[REDACTED: PII detected]"
            record.args = ()
        return True

logger = logging.getLogger(__name__)
logger.addFilter(PiiFilter())

This is a blunt instrument. A better approach is structured logging where your log payload is the same labeled dictionary, and the formatter redacts based on the sensitivity tag rather than regex.

Static analysis as a backstop

Runtime labeling is powerful but it requires discipline. Someone will forget to wrap a new field, or they will cast a Labeled value back to a primitive to avoid a type error.

Static analysis fills the gap. Tools like Semgrep or custom linters can enforce that raw strings from HTTP request bodies are never passed directly to loggers or database sinks without passing through a labeling constructor.

A minimal Semgrep rule might look like this:

rules:
  - id: unlabeled-pii-log
    patterns:
      - pattern: logger.info($X)
      - metavariable-pattern:
          metavariable: $X
          pattern-either:
            - pattern: request.json[$FIELD]
            - pattern: request.form[$FIELD]
    message: "Logging raw request data without sensitivity labeling"
    languages: [python]
    severity: ERROR

This will not catch every leak, but it will catch the obvious ones, which is where most leaks come from.

The trade-offs nobody wants to talk about

This approach adds friction. Every data transformation now involves an extra metadata field. Serialization payloads get larger. Code reviews become arguments about whether user_agent is PII. (It usually is, by the way. The EU thinks so.)

Performance is a real concern. If you are processing millions of events per second, allocating a Labeled wrapper for every field is measurable overhead. In hot paths, you may need to batch label checks or move enforcement to the edge.

There is also the maintenance burden. Sensitivity labels are only as good as the humans who maintain them. When privacy regulations change, or when your business expands into a new jurisdiction, you need to retag data. There is no magic here. It is work.

What we did not choose: regex scanning at rest

Some teams solve this by scanning data stores after the fact with regex, looking for patterns that look like emails or social security numbers. This is better than nothing, but it is fundamentally reactive.

Regex misses obfuscated data, hashed identifiers, and composite PII. It also produces false positives that erode trust in the system. We considered this approach early on and rejected it because it treats the symptom (data in the wrong place) rather than the cause (data leaving the service unlabeled).

Start with the choke points

You do not need to label every field in every service on day one. Start with the boundaries. Label data as it enters your system from users, third-party APIs, and event streams. Then add enforcement at your highest-risk sinks: analytics pipelines, logging infrastructure, and external integrations.

Once those are in place, expand inward. The goal is not perfect coverage. The goal is making PII leaks expensive to create by accident, and obvious to catch in code review.

If you are building this in Python, the Labeled class above is enough to get started. In TypeScript, a branded type or a similar wrapper works the same way. In Go, you will want a struct with an unexported field to prevent accidental casting.

The tools are simple. The hard part is deciding that untracked PII is a bug, and treating it like one.