← Back to the section

Imagine a developer adds logger.info("User registered: email=%s", user.email) — just one line. Six months later the log aggregator is caught in a breach, and users' email addresses end up in the open. That is what a typical PII leak looks like — one innocent line that later costs the company a fine and lost trust.

Let's look at what PII is, why this data must not appear in logs, responses, and message queues, and how to correctly store an application's secrets.

What PII is and why it matters

PII (Personally Identifiable Information) is data that can identify a specific person: email, phone, full name, address, passport details, IP address, biometrics.

Regulators (GDPR, 152-FZ, and others) treat PII as protected data. A leak through a log, an HTTP response, or a message queue is a security incident with all the consequences: an audit, a fine, notifying users.

The main rule: PII must not leave the layer that needs it to work. Logs, HTTP error responses, and events in a queue are not that place.

PII in logs

Here is a common mistake made in the first days of working with a new service:

import logging

logger = logging.getLogger(__name__)

# Never do this
logger.info("User registered: email=%s phone=%s", customer.email, customer.phone)

# Correct — only the internal identifier
logger.info("User registered: customer_id=%s", customer.id)

Logs go to aggregators (ELK, Datadog, Loki), are often accessible to a wide circle of people on the team, and are sometimes exported to third-party analytics systems. Personal data accumulates there unnoticed.

If diagnostics really do need something from the user, use masking — show only part of the value:

# domain/pii_masking.py

def mask_email(email: str) -> str:
    if not email or "@" not in email:
        return "***"
    local, domain = email.split("@", 1)
    return local[0] + "***@" + domain


def mask_phone(phone: str) -> str:
    if not phone or len(phone) < 4:
        return "***"
    return "***" + phone[-4:]

# Acceptable for diagnostics
logger.info("Email verification sent: customer_id=%s email_mask=%s",
            customer.id, mask_email(customer.email))   # u***@example.com

__str__ and __repr__ of objects with PII

Another hidden trap: if a class with personal data does not override __str__ and __repr__, then logger.info("%s", customer) will automatically print all the fields.

from dataclasses import dataclass

@dataclass
class Customer:
    id: str
    email: str
    phone: str
    full_name: str

    def __repr__(self) -> str:
        return f"Customer(id={self.id!r})"

    def __str__(self) -> str:
        return f"Customer[id={self.id}]"

Now accidental logging of the object will not reveal personal data.

Structured logs (structlog)

If you use structlog, watch what ends up in the request context:

import structlog

log = structlog.get_logger()

# Correct
log.info("order_created", order_id=order.id, customer_id=order.customer_id)

# Incorrect
log.info("order_created", order_id=order.id, customer_email=customer.email)

PII in exception text

Another common mistake is including a PII value directly in the exception text:

# Never do this
raise ValueError(f"Email {email} is invalid format")

What happens to this text afterwards:

  • logger.exception(...) writes str(exc) to the log — and the email ends up there.
  • An exception handler returns detail=str(exc) to the client — the user (or an attacker) sees confirmation of the address.

The correct approach is domain exceptions with an error code, without a PII value:

# domain/customer/errors.py

class CustomerDomainError(Exception):
    def __init__(self, code: str, message: str) -> None:
        self.code = code
        self.message = message
        super().__init__(message)


class InvalidEmailFormatError(CustomerDomainError):
    def __init__(self) -> None:
        super().__init__(
            code="INVALID_EMAIL_FORMAT",
            message="Provided email is in invalid format",
        )

The message "Provided email is in invalid format" describes the problem but does not name the address itself. If diagnostics are needed — a separate log with mask_email(email), not through the exception.

The exception handler in FastAPI

The exception handler is a place where it is especially easy to accidentally "slip" PII into the response to the client.

A typical mistake:

# Never do this
return JSONResponse(
    status_code=400,
    content={"detail": str(exc)},              # may contain PII
)

return JSONResponse(
    status_code=500,
    content={"detail": str(exc.__cause__)},    # internal system details
)

return JSONResponse(
    status_code=500,
    content={"traceback": traceback.format_exc()},  # code structure exposed
)

The correct way is an explicit mapping of the error code to a safe message:

# adapters/in/http/exception_handlers.py
from fastapi import Request
from fastapi.responses import JSONResponse
from domain.order.errors import OrderDomainError

async def order_domain_exception_handler(
    request: Request,
    exc: OrderDomainError,
) -> JSONResponse:
    detail = {
        "ORDER_NOT_FOUND": "Order with given id not found",
        "ORDER_NOT_CANCELLABLE": "Order in current status cannot be cancelled",
    }.get(exc.code, "Order operation failed")

    return JSONResponse(
        status_code=400,
        content={
            "type": "urn:order:domain",
            "title": "Order operation failed",
            "detail": detail,
            "errorCode": exc.code,
        },
    )


async def generic_exception_handler(
    request: Request,
    exc: Exception,
) -> JSONResponse:
    request_id = request.headers.get("X-Request-Id", "unknown")
    return JSONResponse(
        status_code=500,
        content={
            "title": "Internal server error",
            "detail": f"An unexpected error occurred. Reference: {request_id}",
        },
    )

Registration in the application:

from fastapi import FastAPI
from domain.order.errors import OrderDomainError
from adapters.in.http.exception_handlers import (
    order_domain_exception_handler,
    generic_exception_handler,
)

app = FastAPI()
app.add_exception_handler(OrderDomainError, order_domain_exception_handler)
app.add_exception_handler(Exception, generic_exception_handler)

In the case of an unexpected error, the client receives a request_id — with it you can find the details in the logs inside the system without exposing them to the outside.

PII in Kafka events

Kafka is a broadcast channel. All subscribers, including those that appear a year from now, will see every message. If PII ends up there — it is stored in the topic until the retention expires and is accessible to any consumer in the system.

# Bad — all consumers see personal data
@dataclass(frozen=True)
class OrderConfirmedEvent:
    order_id: str
    customer_email: str   # leak
    customer_phone: str   # leak
    total_amount: Decimal


# Correct — only the identifier
@dataclass(frozen=True)
class OrderConfirmedEvent:
    order_id: str
    customer_id: str
    total_amount: Decimal

If the notification service needs the email to send a message, it requests it directly from the customer service:

# adapters/out/http/customer_client.py
import httpx
from settings import Settings

class CustomerClient:
    def __init__(self, settings: Settings, http: httpx.AsyncClient) -> None:
        self._base = settings.customer_service_url
        self._http = http

    async def get_email(self, customer_id: str) -> str:
        response = await self._http.get(
            f"{self._base}/customers/{customer_id}/email",
        )
        response.raise_for_status()
        return response.json()["email"]

This gives targeted access with an audit log on the customer service side — it is visible who requested the email and when.

Secrets not in code and not in git

Database passwords, API keys, client secrets — they must not be stored in code or in .env files committed to git.

# Never do this — a secret directly in the code
DB_PASSWORD = "super-secret-password-prod"

The correct approach is pydantic-settings reading values from environment variables:

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

    db_password: str
    client_secret: str
    jwks_uri: str
    audience: str
    issuer: str

The .env file with the real values is only local, in .gitignore:

.env
.env.local
.env.prod
*.pem
*.key
secrets.yaml

Where to get the values in production:

  1. HashiCorp Vault — vault-secrets-operator or the hvac Python client at startup.
  2. SealedSecrets (Kubernetes) — an encrypted secret in git, decrypted by an operator in the cluster.
  3. Cloud Secret Manager (AWS Secrets Manager, GCP Secret Manager) — the pod's IAM role retrieves the secret directly.
  4. Kubernetes Secret — the minimal option to start with.

Protection against an accidental commit

Even with careful work, an accidental commit of a secret happens. detect-secrets finds secret-like strings before they get into the history:

# .pre-commit-config.yaml
repos:
  - repo: https://github.com/Yelp/detect-secrets
    rev: v1.4.0
    hooks:
      - id: detect-secrets
        args: ["--baseline", ".secrets.baseline"]

If a secret did end up in the git history — it must be changed immediately. Removing the commit does not help: it may be in forks, the CI cache, or already downloaded by someone.

In short

  • PII — email, phone, full name, address, IP — must not be written to logs at any level (including DEBUG).
  • For diagnostics use masking (mask_email, mask_phone), not the address itself.
  • Override __str__ and __repr__ on classes with personal data — otherwise accidental logging of the object will reveal everything.
  • Exceptions must not contain PII values — only an error code and a general message.
  • The exception handler in FastAPI returns predefined text based on the error code, not str(exc) or a stack trace.
  • Kafka events carry only the identifier; the service that needs the details requests them directly.
  • Secrets in code and in git are unacceptable; pydantic-settings + environment variables + Vault/SealedSecrets/Cloud SM.
  • If a secret got into the git history — change it immediately, even if the commit was removed.

Further reading