Writing a freshness check for S04

Description

This is the module's central lesson: it writes check_freshness(), the first file-level function this entire guide runs, and runs it for real on orders_2026-08-14.csv. The result is no surprise — module 1's lesson 5 already anticipated it with evidence —, but it is the first time that result comes out of a reusable function, with Polars, instead of a one-off script with hand-done arithmetic.

Connection to the module. Lessons 2 and 3 built the two arguments this lesson brings together: freshness gets measured over the whole file (lesson 2), using a fixed time reference, never the real clock (lesson 3). check_freshness(df, run_at, sla_hours) is the exact synthesis of those two arguments, turned into code any Kiosko pipeline could import and reuse.

An analogy: from the cashier reading the label by hand, to the automatic reader

Think of two ways to check a product's expiration date in a store. The first: a cashier, every time someone buys something, picks up the product, finds the date printed on the label, reads it, and eyeballs it against today's date. It works, but it depends on that specific cashier remembering to check, and reading the date correctly every time — exactly what module 1's lesson 5's script did: a calculation done by hand, once, for one specific case. The second way: an automatic reader at the register, which scans the barcode, extracts the expiration date encoded in it, compares it against the system's date, and shows a signal — green or red — with no cashier having to read or calculate anything.

check_freshness() is that automatic reader. It doesn't replace module 1, lesson 5's knowledge — the same idea of comparing a date against a limit —, it turns it into a piece you can call on any DataFrame with a time column, with nobody having to rewrite the arithmetic every time. And, like any well-designed automatic reader, it works the same no matter which product (or which file) you put in front of it — today's lesson tests it first with toy data, then with S04's real file.

Worked example: build, test clean, run on S04

Step 1 — the function

# checks.py
from datetime import datetime

import duckdb
import polars as pl

PIPELINE_RUN_AT = "2026-08-16T09:00:00"


def check_freshness(df: pl.DataFrame, run_at: str, sla_hours: int, timestamp_col: str = "order_ts") -> dict:
    """Compares the DataFrame's most recent order_ts against run_at. Fails if it exceeds sla_hours."""
    latest_ts = df.select(pl.col(timestamp_col).max()).item()
    run_at_dt = datetime.fromisoformat(run_at)
    hours_since_latest = (run_at_dt - latest_ts).total_seconds() / 3600
    return {
        "check": "freshness",
        "latest_row_ts": str(latest_ts),
        "run_at": run_at,
        "sla_hours": sla_hours,
        "hours_since_latest": round(hours_since_latest, 2),
        "status": "PASS" if hours_since_latest <= sla_hours else "FAIL",
    }

Five lines of body, each with a precise role. df.select(pl.col(timestamp_col).max()).item() is lesson 2's direct translation: instead of walking through row by row, a single Polars expression finds the most recent order_ts across the whole DataFrame, and .item() extracts it as a plain Python value (a datetime.datetime), not a one-cell DataFrame. datetime.fromisoformat(run_at) converts the text string the function receives — never datetime.now(), lesson 3's whole rule — into a comparable object. The subtraction and division by 3600 are the same hour arithmetic module 1's freshness_preview.py already used. And the output dict isn't a style whim: every field documents, right inside the result, what data the verdict got calculated with — latest_row_ts, run_at, sla_hours — with no one reading it having to guess or go back to the source code.

Step 2 — test it with toy data, before touching S04

Before trusting this function with the real file, confirm it with a minimal, hand-built case, where you already know beforehand what the answer should be:

# checks.py -- continuation
if __name__ == "__main__":
    from datetime import datetime as dt

    toy_df = pl.DataFrame({
        "order_id": ["T1", "T2"],
        "order_ts": [dt(2026, 1, 1, 10, 0, 0), dt(2026, 1, 1, 12, 0, 0)],
    })

    print("=== check_freshness on toy data ===")
    print("PASS case (2 hours of delay, 24-hour SLA):")
    print(f"  {check_freshness(toy_df, run_at='2026-01-01T14:00:00', sla_hours=24)}")

    print("\nFAIL case (50 hours of delay, 24-hour SLA):")
    print(f"  {check_freshness(toy_df, run_at='2026-01-03T14:00:00', sla_hours=24)}")

What to expect. Running python3 checks.py up to this point, the output is exactly this:

=== check_freshness on toy data ===
PASS case (2 hours of delay, 24-hour SLA):
  {'check': 'freshness', 'latest_row_ts': '2026-01-01 12:00:00', 'run_at': '2026-01-01T14:00:00', 'sla_hours': 24, 'hours_since_latest': 2.0, 'status': 'PASS'}

FAIL case (50 hours of delay, 24-hour SLA):
  {'check': 'freshness', 'latest_row_ts': '2026-01-01 12:00:00', 'run_at': '2026-01-03T14:00:00', 'sla_hours': 24, 'hours_since_latest': 50.0, 'status': 'FAIL'}

Two toy rows, T1 at 10:00 and T2 at 12:00 — latest_row_ts picks the more recent of the two (12:00:00), exactly as lesson 2 requires: it doesn't matter how many rows the DataFrame has, only the newest one matters. With run_at two hours after that most recent row, PASS; with run_at two days after, FAIL. The function behaved exactly as expected in both extreme cases, before risking any conclusion about S04's real file.

Step 3 — run for real on orders_2026-08-14.csv

# checks.py -- continuation
    con = duckdb.connect("kiosko.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE orders_s04 AS
        SELECT * FROM read_csv('orders_2026-08-14.csv', header=True,
            columns={
                'order_id': 'VARCHAR', 'store_id': 'VARCHAR', 'product_id': 'VARCHAR',
                'quantity': 'BIGINT', 'unit_price': 'DOUBLE', 'order_ts': 'TIMESTAMP'
            })
    """)
    df = con.sql("SELECT * FROM orders_s04").pl()
    print(f"\norders_s04: {df.height} rows\n")

    freshness_result = check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=24)
    print("=== check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=24) ===")
    for k, v in freshness_result.items():
        print(f"  {k}: {v}")

What to expect. With orders_2026-08-14.csv (module 1's exact twelve lines) in the same folder, the additional output is exactly this:

orders_s04: 12 rows

=== check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=24) ===
  check: freshness
  latest_row_ts: 2026-08-14 09:25:00
  run_at: 2026-08-16T09:00:00
  sla_hours: 24
  hours_since_latest: 47.58
  status: FAIL

There it is: the first time, in this entire guide, a file-level check actually runs and fails — not calculated in prose, not a prediction, a real dict, returned by a real function. latest_row_ts: 2026-08-14 09:25:00 is ORD-9511, the latest sale of the twelve the file brings. hours_since_latest: 47.58 is the time elapsed between that sale and PIPELINE_RUN_AT, almost double the agreed 24-hour SLA. status: FAIL.

Why this number (47.58) isn't the same as module 1, lesson 5's (33.0)

If you compared this result against module 1's freshness_preview.py's, you're going to notice a real difference: that script reported 33.0 hours over the SLA; check_freshness() reports 47.58 hours since the most recent row. Both numbers are correct — they measure different questions, on purpose. Module 1's script compared PIPELINE_RUN_AT against SLA_DEADLINE (EXPECTED_ARRIVAL + 24 hours = 2026-08-15T00:00:00), a date calculated beforehand, based on when someone at Kiosko expected the file to arrive — data only a human, with business context, could know in advance. check_freshness(), by contrast, doesn't need anyone to tell it beforehand "when it was expected": it just looks at the data the file actually brings, finds the most recent (09:25:00 on August 14), and compares it directly against run_at.

This difference isn't an error — it's, precisely, what makes check_freshness() reusable beyond S04. Module 1's calculation only works if someone maintains, by hand, a calendar of "when each file is expected" for each Kiosko store. check_freshness() works on any table with a time column, with no such calendar — the same definition, independently verified, that dbt source freshness uses in production: comparing the most recent available data against the review moment, with no need to know beforehand when it "should" have arrived. The two numbers — 33.0 and 47.58 — tell the same story (the file arrived late), with two different, both legitimate, measuring sticks.

Diagram: from module 1's lesson 1 to this lesson

flowchart LR
    A["Module 1, L5:\nfreshness_preview.py\nfixed EXPECTED_ARRIVAL + SLA\nby hand, one-off use"] --> B["33.0 hours over the SLA\n(prose, not reusable)"]
    C["Module 6, L2-L3:\nfreshness is file-level,\nthe clock has to be fixed"] --> D["check_freshness(df, run_at, sla_hours)\nreal, reusable function"]
    D --> E["Tested on toy data:\nPASS and FAIL, both correct"]
    E --> F["Run on real S04:\nlatest_row_ts=09:25:00\nhours_since_latest=47.58\nFAIL"]

Going deeper: what this check "failing" means, in practice

It's worth being precise about what it implies, at this point in the guide, for check_freshness() to return status: FAIL. It doesn't mean the file gets discarded, or that its twelve rows stop being processed — that would repeat foundations M7's error, the total rejection module 1's lesson 5 already criticized. It means, precisely, that Kiosko now has structured evidence — a dict, not a feeling — that this file violated a punctuality agreement, available for a later decision system (this guide's module 7, with quarantine() and raise_alert()) to decide what to do with that information. check_freshness(), on its own, only detects and reports — the same division of responsibility that has already sustained every one of this guide's five earlier tools: a function that decides "does it pass or not?", separate from another function that decides "and now what do we do about it?".

Common mistakes

Using .min() instead of .max() to find the reference order_ts. What happens: someone, without thinking carefully, uses pl.col(timestamp_col).min() instead of .max(), and gets a much worse freshness result than the real one (comparing against the file's oldest sale, not the most recent). Why it happens: min/max are syntactically interchangeable, and without connecting the code to the business question, it's easy to pick the wrong one by a typing slip. How to spot it: if your latest_row_ts is the file's oldest row (for S04, it would be ORD-9501 at 08:05:00, not ORD-9511 at 09:25:00), you flipped the criterion. How to fix it: remember the exact question freshness answers — "how up to date is the file relative to its newest data?" — always .max(), never .min(), on the time column.

Forgetting run_at is an ISO text string, and passing it a datetime object directly. What happens: someone calls check_freshness(df, run_at=datetime(2026, 8, 16, 9, 0, 0), sla_hours=24), passing a datetime object instead of the "2026-08-16T09:00:00" string the function expects, and gets an AttributeError inside datetime.fromisoformat(run_at) (which expects a string, not another datetime). Why it happens: PIPELINE_RUN_AT looks, at a glance, like it could be either type, and the function doesn't explicitly validate which one it received. How to spot it: if the error mentions fromisoformat and the argument you passed isn't a text string, check your run_at's exact type. How to fix it: respect this lesson's exact signature — run_at: str, always a string in ISO 8601 format, exactly as PIPELINE_RUN_AT has been defined since module 1. If in your own code you prefer working directly with datetime objects, this module's lesson 3, exercise 2 already showed how to adapt the function for that case.

Confusing sla_hours=24 with "the file has 24 hours to arrive from when it's generated." What happens: someone interprets the sla_hours parameter as if it described the total time a file can take between being generated and arriving at Kiosko — when it actually describes something more specific: how much time can pass between the most recent data inside the file and the moment someone reviews it. Why it happens: the two interpretations sound similar in loose prose. How to spot it: review exactly what hours_since_latest compares — run_at (when it was reviewed) against latest_row_ts (the most recent sale the file contains), not against any concept of "when the file, as an object, was generated" (a CSV has no timestamp of its own beyond the data it contains). How to fix it: always think of sla_hours as "how much time, at most, can pass between the most recent sale we know about and the moment we decide to review the file" — the exact definition this lesson uses, and the same one dbt source freshness uses in production.

Exercises

Exercise 1 — Run check_freshness() with a much looser sla_hours, and confirm S04 would pass. Using S04's real df, call check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=72) (a three-day SLA instead of one). Does it pass or fail?

loose_result = check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=72)
print(loose_result)

{'check': 'freshness', 'latest_row_ts': '2026-08-14 09:25:00', 'run_at': '2026-08-16T09:00:00', 'sla_hours': 72, 'hours_since_latest': 47.58, 'status': 'PASS'}

Exercise 2 — Confirm that the DataFrame's row order doesn't affect the result. Reload orders_s04 with SELECT * FROM orders_s04 ORDER BY order_id (alphabetical order by order_id, different from the CSV's original order) and run check_freshness() again. Does any result value change?

reordered_df = con.sql("SELECT * FROM orders_s04 ORDER BY order_id").pl()
print(check_freshness(reordered_df, run_at=PIPELINE_RUN_AT, sla_hours=24))

Exercise 3 — Argue whether check_freshness() would keep working correctly if orders_s04 had a row with a null order_ts. In 2-3 sentences, considering how pl.col(timestamp_col).max() behaves against null values in Polars, explain what would happen if, say, ORD-9503 also had an empty order_ts (not just its already-empty unit_price).

Summary and next step

In this lesson you wrote check_freshness(), the direct synthesis of lessons 2 and 3's two arguments: a function that measures the whole file (using the maximum order_ts, never row by row) against a fixed time reference (run_at, never the real clock). You tested it first on toy data, confirming it correctly produces both PASS and FAIL in cases where you already knew the answer beforehand, and you finally ran it on orders_2026-08-14.csv for real: latest_row_ts: 2026-08-14 09:25:00, hours_since_latest: 47.58, status: FAIL — the sixth and last of this guide's six data quality dimensions, with executed evidence.

Before moving on you should be able to: explain, without looking at the code, why check_freshness() uses .max() on the time column; reproduce this lesson's FAIL result by running the code yourself; and explain why its number (47.58 hours) differs from, yet is also correct alongside, the 33.0 module 1, lesson 5 calculated.

Freshness already has its check. One more piece remains before closing the file-level diagnosis: lesson 5 builds check_volume() — and, unlike this lesson, the result on S04 is going to be PASS, the deliberate contrast that not everything in this file is broken.

Resources

• Polars — official documentation, aggregation expressions (max, min, and their behavior against null values, relevant to this lesson's Exercise 3). docs.pola.rs. In English.
• dbt Labs — official documentation, "Add freshness checks to sources" (the independently verified freshness definition that confirms check_freshness()'s design: comparing the most recent available timestamp against the review moment). docs.getdbt.com/reference/resource-properties/freshness. In English.
• Module 1, lesson 5, of this same guide — the source of freshness_preview.py and the 33.0-hour number, contrasted in this lesson against 47.58. src/guides/data-reliability-and-governance-guide/workbook/module-01-when-green-does-not-mean-correct/en/05-meet-s04-kioskos-fourth-store.md. In English.
• Module 2, lesson 4, of this same guide — the exact source of the CREATE OR REPLACE TABLE orders_s04 + .pl() bridge reused in this lesson. src/guides/data-reliability-and-governance-guide/workbook/module-02-declarative-data-quality-tests-with-pandera/en/04-installing-pandera-and-bridging-duckdb-to-polars.md. In English.
• This guide's DESIGN — check_freshness(df, run_at=PIPELINE_RUN_AT, sla_hours=24)'s exact mandate. src/guides/data-reliability-and-governance-guide/DISENO.md. In Spanish.