Capstone: a validated, duplicate-aware catalogue report

pythonscriptingdata-boundariesindependent-practice

Saved on this device

Full lesson · 4 learning steps Open any exercise for focused practice

Step 1 of 4 · intro card

Capstone: a validated, duplicate-aware catalogue report

Concept and contract

Build a small reproducible pipeline that preserves identity, rejects conflicts and counts only the selected population.

The capstone accepts unknown record-like input and a requested country, normalizes declared string fields, requires an actual boolean active flag and produces a small report. It uses only fictional local observations. No production feed, vacancy, profile or database is changed by this exercise.

Repeated observations with the same canonical identity and equal normalized content count once. If the same identity has conflicting canonical content, the function raises ValueError rather than choosing whichever observation happened to arrive last. That explicit conflict policy prevents a convenient report from silently hiding contradictory evidence.

Mechanism and worked example

The pipeline validates and normalizes each record before constructing an identity mapping.

The pipeline validates and normalizes each record before constructing an identity mapping. It checks conflicts across the full validated population before applying the active-country view. A conflict does not become acceptable merely because the current requested country would exclude one of its observations.

The selected stage then filters unique canonical records, sorts their identities and counts distinct company keys. These measures share the same eligible population. The returned company count is not the number of raw source observations or a global company-directory size.

Complete worked example

python

def summarize(rows, country):
    if not isinstance(rows, list) or not isinstance(country, str) or not country.strip():
        raise ValueError("Expected records and a country")
    fields = ("id", "company", "country")
    by_id = {}
    for row in rows:
        if not isinstance(row, dict) or type(row.get("active")) is not bool:
            raise ValueError("Expected a boolean active field")
        if not all(isinstance(row.get(key), str) and row[key].strip() for key in fields):
            raise ValueError("Expected nonempty fields")
        clean = {key: row[key].strip() for key in fields}
        clean["country"] = clean["country"].upper()
        clean["active"] = row["active"]
        identity = clean["id"]
        if identity in by_id and by_id[identity] != clean:
            raise ValueError("Conflicting identity")
        by_id[identity] = clean
    selected = [row for row in by_id.values()
                if row["active"] and row["country"] == country.strip().upper()]
    ids = sorted(row["id"] for row in selected)
    companies = len({row["company"] for row in selected})
    return {"ids": ids, "companies": companies}

Reasoning: Canonical conflicts are checked before the selected view; equal duplicates count once.

Counterexamples and common mistakes

The counterexample overwrites matching identity keys without checking content.

The counterexample overwrites matching identity keys without checking content. It can appear correct when duplicates are identical, while a conflicting country or status silently changes the report. A title-only deduplication policy would create another problem: different legitimate role identities can share the same title.

The teaching key and canonical fields are deliberately small. Real feed reconciliation needs source provenance, stable business identities and richer conflict-resolution evidence. This exercise demonstrates an honest boundary and conflict signal rather than claiming to implement every production merge policy.

Counterexample: a plausible but incorrect implementation

python

def summarize(rows, country):
    if not isinstance(rows, list) or not isinstance(country, str) or not country.strip():
        raise ValueError("Expected records and a country")
    fields = ("id", "company", "country")
    by_id = {}
    for row in rows:
        if not isinstance(row, dict) or type(row.get("active")) is not bool:
            raise ValueError("Expected a boolean active field")
        if not all(isinstance(row.get(key), str) and row[key].strip() for key in fields):
            raise ValueError("Expected nonempty fields")
        clean = {key: row[key].strip() for key in fields}
        clean["country"] = clean["country"].upper()
        clean["active"] = row["active"]
        identity = clean["id"]
        by_id[identity] = clean
    selected = [row for row in by_id.values()
                if row["active"] and row["country"] == country.strip().upper()]
    ids = sorted(row["id"] for row in selected)
    companies = len({row["company"] for row in selected})
    return {"ids": ids, "companies": companies}

Reasoning: Last-observation overwrite hides conflicting canonical facts.

Independent practice: small Python task

Add a count field equal to the number of selected unique role identities, keeping all existing validation, conflict handling and company-count behavior.

Add a count field equal to the number of selected unique role identities, keeping all existing validation, conflict handling and company-count behavior. For the Ukrainian duplicate fixture, predict one selected identity, one company and count one. Do not count the duplicate observation twice.

The independent change belongs in the final report shape, after canonicalization and selection. Reuse the already calculated ids list instead of rebuilding another raw-source count with a slightly different filter. The starter retains the explained pipeline so the task is one focused extension.

Step 1: Keep the canonicalization and selected stage unchanged. Step 2: Add count using the already computed ids list.

Practice starter: solve before reading the solution

python

def summarize(rows, country):
    if not isinstance(rows, list) or not isinstance(country, str) or not country.strip():
        raise ValueError("Expected records and a country")
    fields = ("id", "company", "country")
    by_id = {}
    for row in rows:
        if not isinstance(row, dict) or type(row.get("active")) is not bool:
            raise ValueError("Expected a boolean active field")
        if not all(isinstance(row.get(key), str) and row[key].strip() for key in fields):
            raise ValueError("Expected nonempty fields")
        clean = {key: row[key].strip() for key in fields}
        clean["country"] = clean["country"].upper()
        clean["active"] = row["active"]
        identity = clean["id"]
        if identity in by_id and by_id[identity] != clean:
            raise ValueError("Conflicting identity")
        by_id[identity] = clean
    selected = [row for row in by_id.values()
                if row["active"] and row["country"] == country.strip().upper()]
    ids = sorted(row["id"] for row in selected)
    companies = len({row["company"] for row in selected})
    # Return ids, companies and count from the same selected population.
    raise NotImplementedError

Reasoning: Add one final count using the selected unique identity list.

Hint 1: len(ids) already measures the eligible unique roles.
Hint 2: Do not count raw rows or alter the company count.

Solution, tests and trade-offs

The solution adds len(ids) to the report and preserves every earlier boundary.

The solution adds len(ids) to the report and preserves every earlier boundary. Identical normalized duplicates remain idempotent, conflicting identities still raise and a boolean-looking integer remains invalid. Empty eligible populations return empty identities and zero measures rather than inventing an unknown company.

Acceptance includes normalization, duplicate equivalence, conflicting identities outside the visible country, invalid fields and a no-match result. The caller's input is compared unchanged. These cases establish the small teaching contract without pretending a local function validates real-world source ownership or legal publication rights.

Practice solution and acceptance checks

python

def summarize(rows, country):
    if not isinstance(rows, list) or not isinstance(country, str) or not country.strip():
        raise ValueError("Expected records and a country")
    fields = ("id", "company", "country")
    by_id = {}
    for row in rows:
        if not isinstance(row, dict) or type(row.get("active")) is not bool:
            raise ValueError("Expected a boolean active field")
        if not all(isinstance(row.get(key), str) and row[key].strip() for key in fields):
            raise ValueError("Expected nonempty fields")
        clean = {key: row[key].strip() for key in fields}
        clean["country"] = clean["country"].upper()
        clean["active"] = row["active"]
        identity = clean["id"]
        if identity in by_id and by_id[identity] != clean:
            raise ValueError("Conflicting identity")
        by_id[identity] = clean
    selected = [row for row in by_id.values()
                if row["active"] and row["country"] == country.strip().upper()]
    ids = sorted(row["id"] for row in selected)
    companies = len({row["company"] for row in selected})
    return {"ids": ids, "companies": companies, "count": len(ids)}

Reasoning: All measures retain the same population, conflict policy and non-mutation contract.

Interview reasoning and next step

Present the capstone as validation, canonical identity, conflict detection, eligibility and reporting.

Present the capstone as validation, canonical identity, conflict detection, eligibility and reporting. Explain why every measure is derived from the same selected population and why a conflict is handled before view filtering. This is practical reasoning for data work, test design and interviews.

Keep the solution as a reference after attempting the small extension. The app saves reading and self-check progress, while the companion lab executes the Python files with a selected interpreter. Further Python topics such as async services, advanced packaging and concurrency deserve separate depth rather than being hidden in this first foundation project.

Optional stretch — after the core task works: Design a provenance-bearing conflict report without silently choosing a winner or changing the existing accepted records.

Computer practice: https://applydjinn.com/typorithm/python-lab-20261105.zip. Run verify.py with your selected Python interpreter, then complete an exercise and check it with --practice followed by the lesson id. The lab uses only the standard library and disposable local fixtures; it never connects to the production database or company feeds.

Build a small reproducible pipeline that preserves identity, rejects conflicts and counts only the selected population.