"""VetsLearn Intermediate 03 practice: reference solution. Open this after your own attempt. Standard library only (csv, json, sqlite3). Run it from this folder: python solve.py print the answers as JSON python solve.py --check compare the answers with expected.json python solve.py --sql load the files and run every query in solve.sql python solve.py --pandas OPTIONAL: repeat the counts with pandas, if installed Nothing here edits the CSV files. Cleaning happens in the query. """ import csv import json import sqlite3 import sys from datetime import date from pathlib import Path HERE = Path(__file__).resolve().parent def read_rows(name): with open(HERE / name, newline="", encoding="utf-8") as handle: return list(csv.DictReader(handle)) def load(): """Load both files into an in-memory database. Every column is text.""" con = sqlite3.connect(":memory:") con.execute("CREATE TABLE people (person_id TEXT, name TEXT, branch TEXT)") con.execute( "CREATE TABLE applications (app_id TEXT, person_id TEXT, role TEXT, submitted_on TEXT)" ) # Values are bound with ? placeholders. They are never pasted into the SQL text. con.executemany( "INSERT INTO people VALUES (?, ?, ?)", [(r["person_id"], r["name"], r["branch"]) for r in read_rows("people.csv")], ) con.executemany( "INSERT INTO applications VALUES (?, ?, ?, ?)", [ (r["app_id"], r["person_id"], r["role"], r["submitted_on"]) for r in read_rows("applications.csv") ], ) return con def is_calendar_date(text): try: return date.fromisoformat(text).isoformat() == text except ValueError: return False def one(con, sql, params=()): return con.execute(sql, params).fetchone()[0] def solve(con): people_rows = one(con, "SELECT COUNT(*) FROM people") distinct_people = one(con, "SELECT COUNT(DISTINCT person_id) FROM people") application_rows = one(con, "SELECT COUNT(*) FROM applications") duplicated = [ row[0] for row in con.execute( "SELECT person_id FROM people GROUP BY person_id HAVING COUNT(*) > ? ORDER BY person_id", (1,), ) ] left_join_rows = one( con, "SELECT COUNT(*) FROM people p LEFT JOIN applications a ON a.person_id = p.person_id", ) inner_join_rows = one( con, "SELECT COUNT(*) FROM people p JOIN applications a ON a.person_id = p.person_id", ) per_person = dict( con.execute( "SELECT p.person_id, COUNT(a.app_id) " "FROM (SELECT DISTINCT person_id FROM people) p " "LEFT JOIN applications a ON a.person_id = p.person_id " "GROUP BY p.person_id ORDER BY p.person_id" ) ) orphans = con.execute( "SELECT a.app_id, a.person_id FROM applications a " "LEFT JOIN people p ON p.person_id = a.person_id " "WHERE p.person_id IS NULL ORDER BY a.app_id" ).fetchall() # Python checks the calendar. solve.sql shows the same check in SQL. bad_dates = [ (app_id, submitted_on) for app_id, submitted_on in con.execute( "SELECT app_id, submitted_on FROM applications ORDER BY app_id" ) if not is_calendar_date(submitted_on) ] problems = [] for person_id in duplicated: # A value from the data goes into the query through a placeholder. count = one(con, "SELECT COUNT(*) FROM people WHERE person_id = ?", (person_id,)) problems.append( { "kind": "duplicate person_id", "file": "people.csv", "id": person_id, "why": f"person_id {person_id} is on {count} rows of people.csv, so a join returns that person's applications {count} times", "handling": "count one row per person_id; keep the file as received and report the repeat", } ) for app_id, person_id in orphans: problems.append( { "kind": "no matching person", "file": "applications.csv", "id": app_id, "why": f"{app_id} names person_id {person_id}, which is not in people.csv, so a join from people drops it", "handling": "set aside and report as an application with an unknown owner; do not invent a person", } ) for app_id, submitted_on in bad_dates: problems.append( { "kind": "invalid date", "file": "applications.csv", "id": app_id, "why": f"{app_id} has submitted_on {submitted_on}, which is not a calendar date", "handling": "keep the row, flag it, leave it out of date-based counts; do not guess the day", } ) return { "fixture": "vetslearn-intermediate-03", "revision": "2026-10-09", "people": { "fileRows": people_rows, "distinctPeople": distinct_people, "duplicatedPersonIds": duplicated, }, "applications": { "rows": application_rows, "noMatchingPerson": [{"app_id": a, "person_id": p} for a, p in orphans], "invalidDates": [{"app_id": a, "submitted_on": d} for a, d in bad_dates], }, "join": { "leftJoinRowsAsGiven": left_join_rows, "innerJoinRowsAsGiven": inner_join_rows, "peopleWithAnApplication": sum(1 for n in per_person.values() if n > 0), "peopleWithNoApplication": [p for p, n in per_person.items() if n == 0], }, "applicationsPerPerson": per_person, "problems": problems, "report": { "people": distinct_people, "applications": application_rows, "applicationsWithAKnownOwner": application_rows - len(orphans), }, "lineage": { "files": ["people.csv", "applications.csv"], "filter": "one row per person_id; no application rows removed", "join": "people LEFT JOIN applications ON applications.person_id = people.person_id", "query": "solve.sql and solve.py, revision 2026-10-09", }, } def run_sql_file(con): """Run each statement in solve.sql and print its rows.""" text = (HERE / "solve.sql").read_text(encoding="utf-8") lines = [line for line in text.splitlines() if not line.lstrip().startswith("--")] for statement in "\n".join(lines).split(";"): statement = statement.strip() if not statement: continue cursor = con.execute(statement) print(" | ".join(column[0] for column in cursor.description)) for row in cursor: print(" | ".join("NULL" if value is None else str(value) for value in row)) print() def run_pandas(answers): """OPTIONAL. Needs pandas, which is not part of the standard library.""" try: import pandas as pd except ImportError: print("pandas is not installed. This part is optional; the answers above do not need it.") return 0 people = pd.read_csv(HERE / "people.csv", dtype=str) apps = pd.read_csv(HERE / "applications.csv", dtype=str) joined = people.merge(apps, on="person_id", how="left") parsed = pd.to_datetime(apps["submitted_on"], format="%Y-%m-%d", errors="coerce") figures = { "people file rows": (len(people), answers["people"]["fileRows"]), "distinct people": (people["person_id"].nunique(), answers["people"]["distinctPeople"]), "application rows": (len(apps), answers["applications"]["rows"]), "left join rows": (len(joined), answers["join"]["leftJoinRowsAsGiven"]), "no matching person": ( int((~apps["person_id"].isin(people["person_id"])).sum()), len(answers["applications"]["noMatchingPerson"]), ), "invalid dates": (int(parsed.isna().sum()), len(answers["applications"]["invalidDates"])), } failed = 0 for label, (got, want) in figures.items(): same = int(got) == int(want) failed += 0 if same else 1 print(f"{'same' if same else 'DIFFERENT'}: {label} = {got}") return 1 if failed else 0 def main(argv): con = load() if "--sql" in argv: run_sql_file(con) return 0 answers = solve(con) if "--pandas" in argv: return run_pandas(answers) if "--check" in argv: expected = json.loads((HERE / "expected.json").read_text(encoding="utf-8")) if answers == expected: print("solve.py matches expected.json") return 0 print("solve.py does NOT match expected.json") return 1 print(json.dumps(answers, indent=2)) return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))