Data Quality & Security for FDEs: Validating Input and Protecting PII

Two non-negotiables run through all data work: the data must be trustworthy and it must be protected. Validation is how you keep quality high; a handful of security habits keeps other people's data safe. This tutorial covers both — the gates that stop bad data entering, and the protection of the data you serve.

Table of Contents

• Part 1: Data Quality
• Validation vs Quality
• Where to Validate: The Gate
• Schema Validation with pydantic
• Column Checks in pandas
• Cross-Field and Business Rules
• Quarantine, Don't Delete
• Measuring Data Quality
• Freshness, Completeness, Uniqueness
• Part 2: Data Security
• The Stakes for an FDE
• Secrets: Never in Code
• PII: What You're Protecting
• Masking and Tokenization
• Encryption in Transit and at Rest
• Access Control and Least Privilege
• Auditing and Retention
• Safe Handling in Notebooks and Logs
• Best Practices

Part 1: Data Quality

Validation vs Quality

• Validation = checks that reject or flag data not meeting rules (types, required fields, ranges). It's an action at a point in time.
• Data quality = the measured health of data over time: is it complete, fresh, unique, accurate? Validation is how you protect quality.

The earlier you validate, the cheaper the fix: a rule at ingest costs one function; the same bug in the monthly report costs a reconciliation project.

Where to Validate: The Gate

Validate at the boundary — the moment data enters your system (API response, file upload, row insert). That's the cheapest, most effective gate.

source ──▶ [ VALIDATE ] ──▶ transform ──▶ load ──▶ serve
              reject/quarantine bad input here

Best Practice: validate at ingest, dedupe by stable ID, compute derived fields once, audit changes (the four gates from Solution Design Tutorial 5).

Schema Validation with pydantic

pydantic declares the expected shape of a record and validates it. Perfect for API/JSON data.

from pydantic import BaseModel, field_validator
from typing import Optional

class Order(BaseModel):
    order_id: int
    customer_id: int
    amount: float
    status: str
    email: Optional[str] = None

    @field_validator("amount")
    def must_be_positive(cls, v):
        if v < 0:
            raise ValueError("amount must be >= 0")
        return v

    @field_validator("status")
    def known_status(cls, v):
        if v not in {"paid", "pending", "refunded"}:
            raise ValueError(f"unknown status: {v}")
        return v

Explanation: every field has a type; the validators enforce business rules (amount >= 0, known status). A bad record raises ValidationError instead of polluting your store.

try:
    Order(order_id=1, customer_id=2, amount=-5, status="paid")
except Exception as e:
    print("Rejected:", e)

Column Checks in pandas

For tabular data, assert conditions across the whole frame at ingest.

import pandas as pd

def validate(df: pd.DataFrame) -> pd.DataFrame:
    # Required columns present
    required = {"order_id", "customer_id", "amount", "status"}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(f"Missing columns: {missing}")

    # Type & range checks
    assert (df["amount"] >= 0).all(), "negative amounts found"
    assert df["order_id"].notna().all(), "null order_id"
    assert df["status"].isin(["paid","pending","refunded"]).all(), "bad status"
    return df

Explanation: fail fast — if any row violates a rule, the function raises before loading. This is your ingest gate in one place.

Cross-Field and Business Rules

Some rules span columns and can't be a simple type check.

# refunded orders should have amount recorded, not zeroed silently
bad = df[(df["status"] == "refunded") & (df["amount"].isna())]
assert bad.empty, f"{len(bad)} refunded orders missing amount"

# discount can't exceed amount
assert (df["discount"] <= df["amount"]).all(), "discount > amount"

Explanation: these catch logical errors a schema can't — the contradictions that quietly corrupt a report.

Quarantine, Don't Delete

When a row fails, don't crash the whole run or silently drop it. Quarantine it: set it aside with a reason, load the rest.

valid = df[df.apply(row_is_valid, axis=1)]
bad   = df[~df.index.isin(valid.index)]
bad.to_csv("quarantine/orders_2026-08-19.csv", index=False)
log.warning("Quarantined %d bad rows", len(bad))

Explanation: the good rows still load; bad rows go to a quarantine file with a timestamp for later inspection. Nobody loses data, and the pipeline doesn't die on one bad record.

Measuring Data Quality

Quality is a number, not a feeling. Track simple metrics per load:

quality = {
    "rows": len(df),
    "null_ids": int(df["order_id"].isna().sum()),
    "dups": int(df.duplicated(subset=["order_id"]).sum()),
    "freshness_hrs": float((pd.Timestamp.utcnow() - df["updated_at"].max()).total_seconds()/3600),
}
log.info("Quality: %s", quality)

Freshness, Completeness, Uniqueness

These three catch the most real-world issues:

Best Practice: alert on deviation from baseline, not just absolute thresholds — a table that always has 5% nulls is fine; one that jumps to 40% is an incident.

Part 2: Data Security

The Stakes for an FDE

You'll be handed customer databases, employee records, financials. Mishandling them isn't a bug — it's a liability. The good news: a short list of habits prevents almost every incident.

Note: when in doubt, ask the customer's security contact about their compliance needs (GDPR, HIPAA, SOC 2). Don't guess on regulated data.

Secrets: Never in Code

API keys, DB passwords, and tokens are secrets. They belong in environment variables or a secrets manager — never in source or notebooks.

import os
DB_URL = os.getenv("DW_URL")     # read at runtime, not hardcoded

# .env (gitignored)
DW_URL=postgresql://user:pass@host/db

# .gitignore
.env

Explanation: os.getenv pulls the value at runtime; .gitignore keeps .env out of the repo. A committed secret is a leaked secret — rotate it immediately if that happens. Prefer a secrets manager (AWS Secrets Manager, Vault) over .env for production; same idea, better audit and rotation.

PII: What You're Protecting

PII (Personally Identifiable Information) is anything that identifies a person: names, emails, phone numbers, addresses, national IDs, IP addresses, even combinations like ZIP + birthdate.

Rule of thumb: if you could find or target a specific person from a field (or a few combined), treat it as PII and protect it.

Masking and Tokenization

You often need data without exposing the person. Two techniques:

• Masking: hide part of the value for display/non-prod use.
• Tokenization: replace the value with a random token; only the vault maps token → real value.

def mask_email(email: str) -> str:
    local, domain = email.split("@")
    return f"{local[0]}***@{domain}"      # a***@example.com

def mask_phone(phone: str) -> str:
    return "***-***-" + phone[-4:]        # ***-***-1234

Explanation: masked values keep the shape for testing/demos but can't identify anyone. Use masked copies in dev, staging, and logs.

import secrets
def tokenize(value: str, vault: dict) -> str:
    tok = secrets.token_hex(8)
    vault[tok] = value          # only the vault knows the real value
    return tok

Best Practice: work on masked/tokenized data wherever possible; keep raw PII only where the job truly needs it, and minimize who can read it.

Encryption in Transit and at Rest

• In transit: always use TLS/HTTPS. Any http:// call with customer data is exposed on the wire.
• At rest: the database or object store should encrypt stored data. Managed services (Postgres on RDS, S3) do this by default — enable it and keep the keys in a KMS.

requests.get("https://api.example.com/data", headers=headers)  # always https://

Note: encryption protects data from being read if storage is stolen. It does not control who can query it — that's access control (next).

Access Control and Least Privilege

Give every process and person the minimum access they need — no more.

• A pipeline that only reads orders shouldn't have write access to customers.
• A dashboard user shouldn't have DB admin.
• Use a read-only role for analytics; a separate writer role for the ETL.

CREATE ROLE etl_reader LOGIN PASSWORD '...';
GRANT SELECT ON orders TO etl_reader;
-- (no GRANT INSERT/UPDATE/DELETE)

Explanation: least privilege means a compromised credential or a buggy script does limited damage.

Auditing and Retention

• Audit: log who accessed or changed sensitive data. "Who changed this customer's record, and when?" should be answerable.
• Retention: don't keep data forever. Delete or anonymize it when the business no longer needs it — required by GDPR and just good hygiene.

CREATE TABLE audit_log (
  id SERIAL PRIMARY KEY,
  table_name TEXT,
  row_id INTEGER,
  changed_by TEXT,
  changed_at TIMESTAMPTZ DEFAULT NOW()
);

Safe Handling in Notebooks and Logs

Notebooks and logs leak secrets constantly.

# BAD — logs the secret
log.info(f"Connecting with {DB_URL}")

# GOOD — redacted
log.info("Connecting to data warehouse")

Best Practice: before sharing any screenshot, export, or log, ask "does this contain a real person's data?" If yes, mask it.

Best Practices

• Validate at the boundary (ingest); use pydantic for records, pandas asserts for frames; quarantine, don't delete.
• Measure quality (completeness, uniqueness, freshness) and alert on deviation from baseline.
• Secrets in env vars or a manager; .gitignore .env; rotate on leak. Never in code or logs.
• Treat anything identifying as PII; mask/tokenize wherever you can.
• HTTPS always; encryption at rest on by default; least privilege per role; audit and set retention.