Skip to content

Python for Data Testing — Language Essentials, Pandas & DataFrames

What this is: the programming companion to the Data Testing guide — just enough Python to be dangerous, then pandas and DataFrames in depth, ending with runnable data-quality test patterns. Same spirit as the TypeScript Cheat Sheet, but for the data side of QA.

1. Typical Python Facts (the 60-Second Orientation)

  • Interpreted & dynamically typed — no compile step; a variable's type is determined at runtime (x = 5 then x = "five" is legal — the flexibility and the danger).
  • Indentation is syntax — blocks are defined by whitespace, not { }. Four spaces is the convention; mixing tabs/spaces is an error.
  • Everything is an object — numbers, strings, functions, even types themselves.
  • Batteries included — a huge standard library (csv, json, datetime, pathlib, sqlite3) before you install anything.
  • The ecosystem is why data people use it — pandas, NumPy, pytest, SQLAlchemy, Great Expectations: the de-facto data & testing stack.
  • Versions: modern code is Python 3.10+; python --version to check. Use a virtual environment per project: python -m venv venv → venv\Scripts\activate (Windows).
  • Zero-indexed, like most languages: items[0] is the first element; items[-1] is the last (negative indexing is idiomatic Python).

2. Core Language Cheat Sheet

2.1 Variables & basic types

count = 42                  # int
price = 19.99               # float
name = "Manoj"              # str
active = True               # bool  (capital T/F)
nothing = None              # Python's null

f"Order {count} costs £{price:.2f}"    # f-string formatting → 'Order 42 costs £19.99'

2.2 The four workhorse collections

# list — ordered, mutable, allows duplicates
statuses = ["placed", "shipped", "delivered"]
statuses.append("cancelled")
statuses[0]          # 'placed'
statuses[-1]         # 'cancelled'
statuses[1:3]        # slice → ['shipped', 'delivered']

# tuple — ordered, IMMUTABLE (great for fixed records)
point = (51.5, -0.12)

# dict — key → value (THE data-testing structure)
row = {"order_id": 1001, "amount": 250.0, "status": "shipped"}
row["amount"]                 # 250.0
row.get("discount", 0)        # 0  (safe access with default — no KeyError)

# set — unique values, fast membership & DIFFERENCE
source_ids = {1, 2, 3, 4, 5}
target_ids = {1, 2, 3, 5}
source_ids - target_ids       # {4}  ← rows LOST in the pipeline
target_ids - source_ids       # set() ← no phantom rows  ✅

2.3 Control flow & functions

def classify(amount: float) -> str:          # type hints = self-documenting tests
    if amount >= 1000:
        return "high"
    elif amount >= 100:
        return "medium"
    return "low"

for row in rows:                              # iterate anything
    print(row["order_id"])

# list comprehension — Python's signature move
amounts   = [r["amount"] for r in rows]
big_rows  = [r for r in rows if r["amount"] > 100]
by_id     = {r["order_id"]: r for r in rows}     # dict comprehension → keyed lookup

2.4 Files, errors, and the test frame

import json, csv
from pathlib import Path

data = json.loads(Path("orders.json").read_text())     # JSON → dicts/lists

try:
    value = float(raw)
except ValueError:                    # catch SPECIFIC exceptions
    quarantine.append(raw)            # bad rows go to quarantine, never vanish

assert len(data) == 84213, f"expected 84213 rows, got {len(data)}"   # the QA verb

pytest in one line: put functions named test_* in a file named test_*.py, use plain assert, run pytest -v. That's the entire entry barrier.

3. Pandas & DataFrames — The Data-Testing Workbench

pandas is Python's data-manipulation library; the DataFrame is its core object — an in-memory table with named, typed columns and fast vectorised operations. For a data tester, the DataFrame is what the browser is for a UI tester: the surface you inspect and assert against.

# pip install pandas
import pandas as pd

3.1 Creating & loading DataFrames

# from a dict (test fixtures)
df = pd.DataFrame({
    "order_id": [1001, 1002, 1003, 1004],
    "customer": ["alice", "bob", "alice", "dana"],
    "amount":   [250.0, 75.5, 1200.0, None],
    "status":   ["shipped", "placed", "delivered", "placed"],
})

# from real sources
df = pd.read_csv("orders.csv")
df = pd.read_parquet("orders.parquet")
df = pd.read_sql("SELECT * FROM orders", connection)
df = pd.read_json("orders.json")

3.2 Inspecting — the first five commands you always run

df.head()          # first 5 rows
df.shape           # (rows, columns) → (4, 4)
df.dtypes          # column types — catches "amount arrived as string"
df.info()          # types + non-null counts in one view
df.describe()      # min/max/mean/quartiles — instant sanity profile

3.3 Selecting & filtering

df["amount"]                          # one column (a Series)
df[["order_id", "amount"]]            # multiple columns
df[df["amount"] > 100]                # boolean filter — rows over 100
df[(df["status"] == "placed") & (df["amount"].isna())]   # combine with & | ~
df.loc[df["customer"] == "alice", "amount"]              # rows by condition, one column
df.iloc[0]                            # first row by position

3.4 Transforming

df["amount_gbp"] = df["amount"] * 0.79                  # new derived column
df["customer"]  = df["customer"].str.upper()            # vectorised string ops
df["order_date"] = pd.to_datetime(df["order_date"])     # proper datetime type
df = df.rename(columns={"customer": "customer_name"})
df = df.sort_values("amount", ascending=False)
df = df.drop_duplicates(subset=["order_id"])            # dedupe by key

3.5 Nulls — the data tester's daily bread

df.isna().sum()                       # null count per column  ← run this constantly
df[df["amount"].isna()]               # SHOW me the offending rows
df["amount"] = df["amount"].fillna(0)         # replace
df = df.dropna(subset=["order_id"])           # or drop rows missing the key

3.6 Grouping & aggregation (reconciliation fuel)

df.groupby("status")["amount"].sum()          # revenue per status
df.groupby("customer").agg(
    orders=("order_id", "count"),
    total =("amount",  "sum"),
    avg   =("amount",  "mean"),
)
df["amount"].sum()                            # grand total → compare to the ledger

3.7 Joining (merge) — and catching join bugs

merged = orders.merge(customers, on="customer_id", how="left", indicator=True)
merged["_merge"].value_counts()
# both          9,950   ← matched
# left_only        50   ← orders whose customer_id has NO customer → referential bug!

indicator=True is the single most useful pandas flag for testers — it turns silent join losses into a countable column.

3.8 Comparing two DataFrames (the regression test)

# strict equality — schema + values (baseline vs new output)
pd.testing.assert_frame_equal(
    baseline.sort_values("order_id").reset_index(drop=True),
    current .sort_values("order_id").reset_index(drop=True),
    check_dtype=False, atol=0.01,          # tolerate float dust
)

# or diff row-by-row
diff = baseline.compare(current)           # shows exactly which cells changed

4. Putting It Together — Runnable Data-Quality Tests

Each test below implements a dimension from the Data Testing guide §8 — drop them into test_orders_quality.py and run pytest -v.

import pandas as pd
import pytest

@pytest.fixture
def df():
    return pd.read_parquet("warehouse/daily_orders.parquet")

# ── Completeness ──────────────────────────────────────────────
def test_row_count_reconciles(df):
    source_count = 84_213                      # from source-system query
    assert len(df) == source_count, f"lost {source_count - len(df)} rows in pipeline"

def test_no_null_keys(df):
    assert df["order_id"].isna().sum() == 0

# ── Uniqueness ────────────────────────────────────────────────
def test_order_id_unique(df):
    dupes = df[df.duplicated(subset=["order_id"], keep=False)]
    assert dupes.empty, f"duplicate keys:\n{dupes[['order_id']].head()}"

# ── Validity ──────────────────────────────────────────────────
def test_status_in_allowed_set(df):
    allowed = {"placed", "shipped", "delivered", "cancelled"}
    bad = set(df["status"].dropna().unique()) - allowed
    assert not bad, f"unexpected status values: {bad}"   # catches new upstream enums!

def test_amounts_non_negative(df):
    assert (df["amount"].dropna() >= 0).all()

# ── Consistency ───────────────────────────────────────────────
def test_totals_match_ledger(df):
    ledger_total = 1_284_550.25
    assert abs(df["amount"].sum() - ledger_total) < 0.01

# ── Referential integrity ─────────────────────────────────────
def test_every_order_has_customer(df):
    customers = pd.read_parquet("warehouse/customers.parquet")
    m = df.merge(customers[["customer_id"]], on="customer_id",
                 how="left", indicator=True)
    orphans = m[m["_merge"] == "left_only"]
    assert orphans.empty, f"{len(orphans)} orders reference missing customers"

# ── Freshness ─────────────────────────────────────────────────
def test_data_is_fresh(df):
    latest = pd.to_datetime(df["loaded_at"]).max()
    assert latest > pd.Timestamp.now() - pd.Timedelta(hours=2)

# ── Snapshot regression (golden baseline) ─────────────────────
def test_matches_golden_baseline():
    baseline = pd.read_parquet("golden/daily_sales_baseline.parquet")
    current  = pd.read_parquet("warehouse/daily_sales.parquet")
    pd.testing.assert_frame_equal(
        baseline.sort_values("region").reset_index(drop=True),
        current .sort_values("region").reset_index(drop=True),
        atol=0.01,
    )

Beyond hand-rolled asserts: when suites grow, graduate to declarative DQ frameworks — Great Expectations or Pandera (schema + checks as config/code), and dbt tests for in-warehouse SQL assertions. The concepts are identical; the pandas versions above are what's happening under the hood.

5. Pandas Quick Reference Card

Task Code
Load pd.read_csv / read_parquet / read_sql / read_json
Peek df.head() · df.shape · df.dtypes · df.describe()
Filter df[df["col"] > x] · & \| ~ for combos
Nulls df.isna().sum() · fillna() · dropna()
Dedupe df.duplicated(subset=[...]) · drop_duplicates()
Group df.groupby("k")["v"].sum() / .agg(...)
Join a.merge(b, on="k", how="left", indicator=True)
Compare pd.testing.assert_frame_equal(a, b) · a.compare(b)
New column df["new"] = df["a"] * 2
Dates pd.to_datetime() · pd.Timestamp.now() · pd.Timedelta

Where to Go Next