Govern FHIR-to-SQL Data Quality
Build a local FHIR-to-SQL validator with privacy-safe quarantine evidence.
Introduction
30 Second Summary
A healthcare dataset can look complete while important details are missing or contradictory. Without visible checks, the next person using it has little reason to trust the result.
In this project, you will build a local data governance and data quality layer for synthetic FHIR-to-SQL exports using Python. One command will turn deliberately flawed records into a quality report plus a privacy-safe quarantine file.
What You'll Build
Run the finished validator to see deliberately flawed Patient data become a five-dimension report plus minimized findings routed by rule, owner, severity, and source row.
By the end of this project, you'll have:
- Edit a reusable rule catalog to test completeness, validity, uniqueness, consistency, and timeliness.
- Run a dependency-free validator to see dimension pass rates, an overall result, and minimized quarantine findings without patient keys or business identifiers.
- Present a governance case study that shows ownership, escalation, privacy controls, limitations, and screenshot-backed evidence.
- Secret Mission: Use the generated evidence to choose a release outcome in a governance tabletop exercise. Defend that decision with named ownership, required remediation, and a re-review condition.
Are there any prerequisites?
You'll need a Mac with Python 3.7 or newer plus Visual Studio Code. An existing FHIR-to-SQL project is optional because this project supplies its own synthetic fixtures.
Before We Start
Before the hands-on work begins, this is your moment to lock in what the quality layer protects and why that protection matters. You are defining the purpose that guides every rule, report, and privacy decision in the project.
Set Up the Quality Workspace
A FHIR-to-SQL quality layer needs its own boundary so an earlier repository stays untouched. This dedicated folder also gives every generated artifact a predictable home.
In this step, you will open that folder in Visual Studio Code. You will confirm that Python meets the minimum version for the validator.
In this step, get ready to:
- Create an isolated workspace for the quality layer.
- Add the folders and empty starter files.
- Confirm Python 3.7 or newer in the integrated terminal.
Create the workspace
Visual Studio Code treats an open folder as a workspace. Creating this one on your Desktop makes it easy to find later.
- Press Cmd+Space on macOS or the Windows key on Windows to open system search.
- Enter Visual Studio Code in the search field.
- Press Enter to launch Visual Studio Code.
You'll see the Visual Studio Code welcome window.
- Select File in the top menu.
- Select Open Folder....
- Select Desktop in the folder dialog sidebar.
- Select New Folder.
- Enter governance-quality-layer as the folder name.
- Press Return to create the folder.
Visual Studio Code may show a Workspace Trust dialog after the folder opens. You created this empty folder yourself, so you can trust it.
- Select Open to open the folder.
- Review the folder name if the Workspace Trust dialog appears.
- Select Yes, I trust the authors if the dialog appears.
You should see governance-quality-layer at the top of the Explorer view. This confirms the isolated workspace is open.
Create the starter structure
The folder structure separates synthetic source data from generated evidence. The starter files reserve clear homes for validation code, quality rules, governance decisions, and the case study.
- Select Terminal in the top menu.
- Select New Terminal.
You'll see the integrated terminal at the bottom of Visual Studio Code. It starts inside the open governance-quality-layer workspace.
- Create the three directories and four empty starter files by running these commands:
mkdir input output docs
touch validate_data.py quality_rules.csv governance_brief.md README.mdWhat do these commands create?
- The mkdir command creates input/, output/, and docs/ inside the workspace.
- The touch command creates the four empty files at the workspace root.
- The empty files establish the exact checkpoint for this setup step.
- Confirm the complete starter structure by running this command:
lsWhat should I see?
The ls command lists the contents of the current workspace.
You should see README.md, docs, governance_brief.md, input, output, quality_rules.csv, and validate_data.py in the output.
Files or folders missing?
- Check that the terminal prompt shows the governance-quality-layer folder before rerunning the creation commands.
- Compare every file name with the command because capitalization matters.
- Still stuck? Help me troubleshoot my workspace structure.
✔️ Awesome, I've got everything!
Great. Your workspace now has a place for every project artifact.
ⓧ I'd like to double check the full code
The four starter files contain no code yet. Their blank state is the exact complete checkpoint for this step.
- The input/ directory exists for synthetic CSV exports.
- The output/ directory exists for generated quality evidence.
- The docs/ directory exists for portfolio evidence screenshots.
- The validate_data.py file exists at the workspace root. Its contents are empty.
- The quality_rules.csv file exists at the workspace root. Its contents are empty.
- The governance_brief.md file exists at the workspace root. Its contents are empty.
- The README.md file exists at the workspace root. Its contents are empty.
Confirm the Python version
The validator relies on ISO date parsing that is available from Python 3.7. This version check protects you from confusing compatibility failures later.
Before you run the check, do you expect your installed Python version to meet the project minimum?
- Confirm the installed Python version in the integrated terminal from earlier by running this command:
python3 --versionWhat does this command do?
The python3 command selects your Python 3 interpreter.
The --version option prints its version number and exits.
✔️ I see version 3.7 or higher
That compatibility check passes. Your Python interpreter can support the validator used in this project.
- Keep the integrated terminal open for the final checkpoint.
ⓧ I see an older version
Your current interpreter is below the project minimum. The official Python installer provides a supported macOS version without removing Apple's managed interpreter.
The installer may request administrator approval. It can take a few minutes while the new interpreter is copied.
- Open the official Python macOS guide.
- Download the current supported macOS installer described in the guide.
- Run the downloaded installer with its default options.
- Return to Visual Studio Code from earlier after the installation finishes.
- Select Terminal in the top menu.
- Select New Terminal to start a fresh terminal session.
- Re-run the version check shown above.
You should now see Python 3.7 or newer.
Still seeing the older version?
- Confirm that you used the default installer options because they make the newer python3 interpreter available to your shell.
- Close any terminal session that was open before the installation.
- Need another set of eyes? Help me use the newer Python installation.
ⓧ Command not found
Your shell cannot currently access the Python 3 interpreter. The official macOS installer adds a supported interpreter for terminal use.
The installation may take a few minutes. Keep Visual Studio Code available so you can return to the same workspace afterward.
- Open the official Python macOS guide.
- Download the current supported macOS installer described in the guide.
- Run the downloaded installer with its default options.
- Return to Visual Studio Code from earlier after the installation finishes.
- Select Terminal in the top menu.
- Select New Terminal to start a fresh terminal session.
- Re-run the version check shown above.
You should now see Python 3.7 or newer.
Still getting no version output?
- Restart Visual Studio Code so its terminal inherits the updated shell configuration.
- Confirm that the Python installer completed successfully before opening a new terminal.
- Still blocked? Help me make Python available in the terminal.
Once the success tab applies, your terminal shows Python 3.7 or newer. The Explorer still lists all seven starter artifacts.
That's the foundation in place. Your isolated workspace is ready for the synthetic exports and first quality check in the next step.
Expose the Baseline Quality Gap
Your Visual Studio Code workspace now gives the quality layer a safe home beside your earlier work. The next question is whether the exported Patient rows deserve downstream trust.
A single data quality rule gives you immediate evidence. Running it also reveals how little one rule can prove about an otherwise plausible dataset.
In this step, get ready to:
- Add synthetic Patient and Organization exports.
- Catalog an enabled completeness rule for the Patient identifier.
- Run a baseline validator to expose the limits of one rule.
Create synthetic input fixtures
These CSV files contain only fictional values. Their columns map selected FHIR elements into a SQL-style export.
Why use synthetic fixtures?
Synthetic fixtures let you test missing identifiers without exposing real clinical data. They also give every learner the same defects to investigate.
Keep these files separate from real exports. The quality workflow starts with safe evidence.
- Create input/patients.csv with the new-file control in the Visual Studio Code Explorer sidebar.
You should now see the empty patients.csv file beneath the existing input folder.
- Fill input/patients.csv by pasting this synthetic Patient export:
patient_key,business_identifier,birth_date,deceased_date,active,managing_organization_key,last_updated
P001,MRN-1001,1984-02-29,,true,ORG-1,2026-10-01T10:00:00+00:00
P002,,2028-01-01,,true,ORG-1,2026-10-02T10:00:00+00:00
P003,MRN-1001,1972-07-10,1971-01-01,false,ORG-404,2024-01-15T00:00:00+00:00
P004,MRN-1004,1992-05-20,,true,ORG-2,2026-09-28T09:30:00+00:00What does this fixture contain?
- The first row defines the SQL-export columns that the validator reads.
- The four data rows provide a small Patient dataset with deliberately varied values.
- The blank business_identifier on the third CSV line creates the baseline completeness failure.
- Save input/patients.csv by pressing Cmd+S on macOS or Ctrl+S on Windows.
You should see the header followed by four Patient rows in the editor.
Patient rows look misaligned?
Check that every row remains on one line. Preserve the empty value between the first two commas on the P002 row.
Still stuck? Help me check the structure of my synthetic Patient CSV fixture.
✔️ Awesome, I've got everything!
Great. Your synthetic Patient fixture is saved inside the existing input folder.
ⓧ I'd like to double check the full code
patient_key,business_identifier,birth_date,deceased_date,active,managing_organization_key,last_updated
P001,MRN-1001,1984-02-29,,true,ORG-1,2026-10-01T10:00:00+00:00
P002,,2028-01-01,,true,ORG-1,2026-10-02T10:00:00+00:00
P003,MRN-1001,1972-07-10,1971-01-01,false,ORG-404,2024-01-15T00:00:00+00:00
P004,MRN-1004,1992-05-20,,true,ORG-2,2026-09-28T09:30:00+00:00Compare the header and all four Patient rows with your saved file. The empty identifier on the P002 row is intentional.
- Create input/organizations.csv with the new-file control in the Visual Studio Code Explorer sidebar.
You should now see the empty organizations.csv file beside patients.csv.
- Fill input/organizations.csv by pasting this synthetic Organization export:
organization_key,organization_name
ORG-1,Synthetic North Clinic
ORG-2,Synthetic South ClinicWhat does this fixture contain?
This fixture defines the Organization keys that a later consistency rule can treat as valid references. The baseline completeness rule does not read this file yet.
Keeping the fixture now makes the unresolved ORG-404 reference available for the expanded validator.
- Save input/organizations.csv by pressing Cmd+S on macOS or Ctrl+S on Windows.
You should see the header followed by the two synthetic Organization rows.
Organization fixture not lining up?
Confirm that the file has exactly two columns. Check that each Organization row contains one comma.
Still stuck? Help me check my synthetic Organization CSV fixture.
✔️ Awesome, I've got everything!
Your two synthetic input fixtures are now ready for validation.
ⓧ I'd like to double check the full code
organization_key,organization_name
ORG-1,Synthetic North Clinic
ORG-2,Synthetic South ClinicCompare the header and both Organization rows with your saved file.
Catalog the first quality rule
A rule catalog turns a governance expectation into structured data that code can execute. This first rule maps Patient.identifier to the exported business_identifier column.
- Switch back to the empty quality_rules.csv file from the workspace setup.
- Add the baseline completeness rule by pasting this content:
rule_id,dimension,resource_type,fhir_element,sql_field,description,severity,threshold_value,owner,escalation_action,enabled
DQ-COMP-001,Completeness,Patient,Patient.identifier,business_identifier,Require a populated business identifier,High,not_blank,FHIR Data Steward,Investigate the source mapping and backfill or reject the row,trueHow does the rule become executable?
- The DQ-COMP-001 identifier gives the check a stable reference for reports and remediation.
- The Completeness dimension states which quality characteristic the rule measures.
- The not_blank threshold defines the condition that business_identifier must satisfy.
- The owner and escalation action route a failure to a named workflow.
- Save quality_rules.csv by pressing Cmd+S on macOS or Ctrl+S on Windows.
You should see one header row followed by the enabled DQ-COMP-001 rule.
Rule split across extra columns?
Keep the description and escalation action free of additional commas. An extra comma changes the number of CSV fields.
Still stuck? Help me validate the structure of my quality rule catalog.
✔️ Awesome, I've got everything!
The catalog now holds one enabled completeness rule with an owner and remediation action.
ⓧ I'd like to double check the full code
rule_id,dimension,resource_type,fhir_element,sql_field,description,severity,threshold_value,owner,escalation_action,enabled
DQ-COMP-001,Completeness,Patient,Patient.identifier,business_identifier,Require a populated business identifier,High,not_blank,FHIR Data Steward,Investigate the source mapping and backfill or reject the row,trueCompare the header and the complete DQ-COMP-001 row with your saved catalog.
Build and test the baseline validator
The Python validator reads the synthetic export and applies the enabled rule. Its first report measures only completeness.
- Switch back to the empty validate_data.py file from the workspace setup.
- Add the paths and CSV-reading foundation by pasting this first chunk:
import csv
from pathlib import Path
BASE_DIR = Path.cwd()
INPUT_DIR = BASE_DIR / "input"
OUTPUT_DIR = BASE_DIR / "output"
PATIENTS_PATH = INPUT_DIR / "patients.csv"
RULES_PATH = BASE_DIR / "quality_rules.csv"
REPORT_PATH = OUTPUT_DIR / "quality_report.csv"
REPORT_FIELDS = [
"dimension",
"total_checks",
"passed_checks",
"failed_checks",
"pass_rate",
"status",
]
def read_csv(path):
with path.open("r", newline="", encoding="utf-8") as handle:
reader = csv.DictReader(handle)
rows = list(reader)
return rows, reader.fieldnames or []What does this code do?
- The path constants anchor every input and output to the open governance-quality-layer workspace.
- The REPORT_FIELDS list defines the aggregate evidence written by the baseline.
- The read_csv() helper returns the file rows plus its column names.
- Save validate_data.py by pressing Cmd+S on macOS or Ctrl+S on Windows.
The editor should show the imports and path constants without syntax warnings.
Seeing syntax warnings already?
Check that each path value uses matching quotation marks. Keep every item in REPORT_FIELDS inside the surrounding square brackets.
Still stuck? Help me inspect the first validator chunk for a Python syntax problem.
- Place the cursor after the final line of read_csv().
- Complete the baseline validator by pasting this second chunk:
def write_csv(path, fieldnames, rows):
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
def check_required_identifier(patients, organizations, run_datetime, rule):
results = []
for source_row, patient in enumerate(patients, start=2):
passed = bool(patient["business_identifier"].strip())
issue = "business_identifier is blank" if not passed else ""
results.append({"source_row": source_row, "passed": passed, "issue": issue})
return results
def main():
patients, patient_fields = read_csv(PATIENTS_PATH)
rules, rule_fields = read_csv(RULES_PATH)
enabled_rules = [rule for rule in rules if rule["enabled"].strip().lower() == "true"]
rule = enabled_rules[0]
evaluations = check_required_identifier(patients, None, None, rule)
failed_checks = sum(1 for evaluation in evaluations if not evaluation["passed"])
report_rows = [{"dimension": rule["dimension"], "total_checks": len(evaluations), "passed_checks": len(evaluations) - failed_checks, "failed_checks": failed_checks, "pass_rate": "{:.1%}".format((len(evaluations) - failed_checks) / len(evaluations)), "status": "PASS" if failed_checks == 0 else "FAIL"}]
write_csv(REPORT_PATH, REPORT_FIELDS, report_rows)
for evaluation in evaluations:
if not evaluation["passed"]:
print("Source row {}: {}".format(evaluation["source_row"], evaluation["issue"]))
if __name__ == "__main__":
main()How does the baseline work?
- The write_csv() helper creates the output directory when needed before writing the report.
- The check_required_identifier() function tests each Patient row for a populated business_identifier.
- Source-row counting starts at 2 because the CSV header occupies the first line.
- The main() routine aggregates the evaluations into one Completeness report row.
- Save validate_data.py by pressing Cmd+S on macOS or Ctrl+S on Windows.
Before you run the validator, do you think one completeness rule will catch every deliberate defect in the Patient fixture?
- Run the baseline validator in the integrated terminal with this command:
python3 validate_data.pyWhat does this command do?
Python executes validate_data.py from the open workspace. The script reads the Patient fixture and the enabled rule catalog.
It writes the aggregate result to output/quality_report.csv before printing each failed source row.
You should see Source row 3: business_identifier is blank in the terminal. That line proves the first rule found its intended defect.
- Select output/quality_report.csv in the Visual Studio Code Explorer sidebar.
You should see a Completeness row with 4 total checks. It shows 3 passed checks and 1 failed check.
The pass rate is 75.0%. The status is FAIL.
What did the baseline miss?
- The duplicate MRN-1001 identifiers remain unchecked.
- The future birth date remains unchecked.
- The death date that precedes the birth date remains unchecked.
- The unresolved ORG-404 reference remains unchecked.
- The stale last_updated value remains unchecked.
That shortfall is intentional. A plausible result from one dimension cannot establish trust across the rest of the dataset.
Baseline report missing?
If Python reports a missing file, confirm that patients.csv sits inside input. Confirm that quality_rules.csv sits at the workspace root.
If the terminal prints no failed row, check that the P002 row still has a blank business_identifier.
Still stuck? Help me troubleshoot my baseline FHIR-to-SQL quality validator.
✔️ Awesome, I've got everything!
Your baseline validator now creates an aggregate completeness report and identifies the blank business identifier on source row 3.
ⓧ I'd like to double check the full code
import csv
from pathlib import Path
BASE_DIR = Path.cwd()
INPUT_DIR = BASE_DIR / "input"
OUTPUT_DIR = BASE_DIR / "output"
PATIENTS_PATH = INPUT_DIR / "patients.csv"
RULES_PATH = BASE_DIR / "quality_rules.csv"
REPORT_PATH = OUTPUT_DIR / "quality_report.csv"
REPORT_FIELDS = [
"dimension",
"total_checks",
"passed_checks",
"failed_checks",
"pass_rate",
"status",
]
def read_csv(path):
with path.open("r", newline="", encoding="utf-8") as handle:
reader = csv.DictReader(handle)
rows = list(reader)
return rows, reader.fieldnames or []
def write_csv(path, fieldnames, rows):
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
def check_required_identifier(patients, organizations, run_datetime, rule):
results = []
for source_row, patient in enumerate(patients, start=2):
passed = bool(patient["business_identifier"].strip())
issue = "business_identifier is blank" if not passed else ""
results.append({"source_row": source_row, "passed": passed, "issue": issue})
return results
def main():
patients, patient_fields = read_csv(PATIENTS_PATH)
rules, rule_fields = read_csv(RULES_PATH)
enabled_rules = [rule for rule in rules if rule["enabled"].strip().lower() == "true"]
rule = enabled_rules[0]
evaluations = check_required_identifier(patients, None, None, rule)
failed_checks = sum(1 for evaluation in evaluations if not evaluation["passed"])
report_rows = [{"dimension": rule["dimension"], "total_checks": len(evaluations), "passed_checks": len(evaluations) - failed_checks, "failed_checks": failed_checks, "pass_rate": "{:.1%}".format((len(evaluations) - failed_checks) / len(evaluations)), "status": "PASS" if failed_checks == 0 else "FAIL"}]
write_csv(REPORT_PATH, REPORT_FIELDS, report_rows)
for evaluation in evaluations:
if not evaluation["passed"]:
print("Source row {}: {}".format(evaluation["source_row"], evaluation["issue"]))
if __name__ == "__main__":
main()Compare the paths and both CSV helpers first. Then check the identifier evaluation and the baseline main() routine.
You have your first executable quality evidence. Next, you will expand the catalog and validator across five quality dimensions so the hidden defects become visible.
Validate Five Quality Dimensions
Your baseline check now catches a blank business identifier. That first result proves the rule catalog can drive an executable quality check.
The same dataset still hides duplicate identifiers, invalid dates, inconsistent chronology, broken organization references, and stale records. This step expands the validator across five quality dimensions so those blind spots become visible.
In this step, get ready to:
- Expand the rule catalog across five data quality dimensions.
- Add evaluators for validity, uniqueness, consistency, referential integrity, and timeliness.
- Generate an aggregate report with dimension-level results.
Expand the rule catalog
A rule catalog turns governance expectations into fields that code can evaluate. Each rule connects a FHIR element to its exported SQL column.
The expanded catalog covers completeness, validity, uniqueness, consistency, and timeliness. It also records severity, ownership, thresholds, and remediation actions.
- In the Visual Studio Code Explorer sidebar, select quality_rules.csv.
- Replace the current catalog with these six enabled rules:
rule_id,dimension,resource_type,fhir_element,sql_field,description,severity,threshold_value,owner,escalation_action,enabled
DQ-COMP-001,Completeness,Patient,Patient.identifier,business_identifier,Require a populated business identifier,High,not_blank,FHIR Data Steward,Investigate the source mapping and backfill or reject the row,true
DQ-VAL-001,Validity,Patient,Patient.birthDate,birth_date,Require a parseable birth date that is not after the run date,High,run_date,FHIR Data Steward,Verify the source value and correct the transformation,true
DQ-UNIQ-001,Uniqueness,Patient,Patient.identifier,business_identifier,Require populated business identifiers to be unique,High,unique,Data Quality Owner,Open a duplicate-record review before release,true
DQ-CONS-001,Consistency,Patient,Patient.deceased[x],deceased_date,Require the deceased date to be blank or not earlier than birth date,Critical,birth_date,Clinical Data Steward,Stop release and verify the source chronology,true
DQ-CONS-002,Consistency,Patient,Patient.managingOrganization,managing_organization_key,Require populated organization references to resolve,High,organizations.csv,Pipeline Engineer,Repair the reference mapping and rerun validation,true
DQ-TIME-001,Timeliness,Patient,Resource.meta.lastUpdated,last_updated,Require a parseable last-updated instant no more than 365 days old,Medium,365,Data Quality Owner,Confirm refresh expectations and reload stale data,trueWhat does this catalog add?
- The rule_id column gives each check a stable identifier that the validator can dispatch.
- The fhir_element and sql_field columns document the mapping from the source standard to the exported column.
- The ownership columns route each failure to a named role with a specific remediation action.
- The enabled column lets the validator select active governance rules.
- Save quality_rules.csv.
- Count the rule rows below the header in the editor. You should see six enabled rules across five dimensions.
Catalog rows not lining up?
- Check that every row contains the same eleven comma-separated fields as the header.
- Confirm that each rule ends with true.
- Still stuck? Help me compare my quality rule catalog with the required CSV structure.
✔️ Awesome, I've got everything!
Your catalog now contains six enabled rules covering five quality dimensions. Save quality_rules.csv before continuing.
ⓧ I'd like to double check the full code
Compare your complete quality_rules.csv file with this reference:
rule_id,dimension,resource_type,fhir_element,sql_field,description,severity,threshold_value,owner,escalation_action,enabled
DQ-COMP-001,Completeness,Patient,Patient.identifier,business_identifier,Require a populated business identifier,High,not_blank,FHIR Data Steward,Investigate the source mapping and backfill or reject the row,true
DQ-VAL-001,Validity,Patient,Patient.birthDate,birth_date,Require a parseable birth date that is not after the run date,High,run_date,FHIR Data Steward,Verify the source value and correct the transformation,true
DQ-UNIQ-001,Uniqueness,Patient,Patient.identifier,business_identifier,Require populated business identifiers to be unique,High,unique,Data Quality Owner,Open a duplicate-record review before release,true
DQ-CONS-001,Consistency,Patient,Patient.deceased[x],deceased_date,Require the deceased date to be blank or not earlier than birth date,Critical,birth_date,Clinical Data Steward,Stop release and verify the source chronology,true
DQ-CONS-002,Consistency,Patient,Patient.managingOrganization,managing_organization_key,Require populated organization references to resolve,High,organizations.csv,Pipeline Engineer,Repair the reference mapping and rerun validation,true
DQ-TIME-001,Timeliness,Patient,Resource.meta.lastUpdated,last_updated,Require a parseable last-updated instant no more than 365 days old,Medium,365,Data Quality Owner,Confirm refresh expectations and reload stale data,trueAdd the five-dimension evaluators
Each rule needs an evaluator that returns the same result shape. A shared shape lets the aggregation routine count outcomes without knowing the details of each check.
- In the Visual Studio Code Explorer sidebar, select validate_data.py.
- Replace the imports and path constants at the top of the file with this expanded configuration:
import csv
import datetime as dt
from pathlib import Path
BASE_DIR = Path.cwd()
INPUT_DIR = BASE_DIR / "input"
OUTPUT_DIR = BASE_DIR / "output"
PATIENTS_PATH = INPUT_DIR / "patients.csv"
ORGANIZATIONS_PATH = INPUT_DIR / "organizations.csv"
RULES_PATH = BASE_DIR / "quality_rules.csv"
REPORT_PATH = OUTPUT_DIR / "quality_report.csv"
QUARANTINE_PATH = OUTPUT_DIR / "quarantined_records.csv"
DIMENSIONS = [
"Completeness",
"Validity",
"Uniqueness",
"Consistency",
"Timeliness",
]
REPORT_FIELDS = [
"dimension",
"total_checks",
"passed_checks",
"failed_checks",
"pass_rate",
"status",
]What does this configuration do?
- The datetime import supports date validity and timeliness checks.
- The ORGANIZATIONS_PATH constant gives the reference check access to the synthetic organization export.
- The DIMENSIONS list fixes the order of the five report rows.
- The REPORT_FIELDS list defines the aggregate evidence written for each dimension.
- Add the quarantine field names immediately below REPORT_FIELDS using this block:
QUARANTINE_FIELDS = [
"quarantine_id",
"resource_type",
"source_row",
"rule_id",
"dimension",
"severity",
"failed_field",
"observed_issue",
"remediation_action",
"owner",
]Why define these fields now?
The validator captures structured failure metadata while it evaluates each rule. The next step uses this schema to inspect the privacy-safe quarantine workflow.
- Keep the existing read_csv and write_csv functions in place.
- Replace the current completeness evaluator with these shared helpers and the updated evaluator:
def require_columns(actual_fields, required_fields, path):
missing = [field for field in required_fields if field not in actual_fields]
if missing:
raise ValueError(
"{} is missing required columns: {}".format(path, ", ".join(missing))
)
def check_result(source_row, passed, issue=""):
return {
"source_row": source_row,
"passed": passed,
"issue": issue,
}
def check_required_identifier(patients, organizations, run_datetime, rule):
results = []
for source_row, patient in enumerate(patients, start=2):
passed = bool(patient["business_identifier"].strip())
issue = "business_identifier is blank" if not passed else ""
results.append(check_result(source_row, passed, issue))
return resultsHow do the shared helpers work?
- The require_columns function stops validation when an input contract is missing a required field.
- The check_result function gives every evaluator the same source row, outcome, and issue structure.
- The updated check_required_identifier function now accepts the same arguments as every other checker.
- Save validate_data.py.
- Confirm the expanded configuration preserves the baseline run by running:
python3 validate_data.pyWhat does this check prove?
The existing baseline routine should still complete and regenerate output/quality_report.csv. This confirms that the shared configuration and helper changes preserve the working completeness check.
Seeing a syntax or name error?
- Check that import datetime as dt appears directly below import csv.
- Confirm that read_csv and write_csv remain below the constants.
- Still stuck? Help me fix the configuration and shared helper changes in my validator.
- Add the birth-date evaluator below check_required_identifier using this function:
def check_birth_date(patients, organizations, run_datetime, rule):
results = []
run_date = run_datetime.date()
for source_row, patient in enumerate(patients, start=2):
value = patient["birth_date"].strip()
if not value:
results.append(check_result(source_row, True))
continue
try:
birth_date = dt.date.fromisoformat(value)
passed = birth_date <= run_date
except ValueError:
passed = False
issue = "birth_date is invalid or after the run date" if not passed else ""
results.append(check_result(source_row, passed, issue))
return resultsHow is birth-date validity tested?
- The evaluator parses populated values as ISO dates.
- It compares each parsed date with the UTC run date.
- It records a generic issue when parsing fails or the birth date falls after the run date.
- Add the identifier uniqueness evaluator directly below check_birth_date using this function:
def check_unique_identifier(patients, organizations, run_datetime, rule):
counts = {}
for patient in patients:
value = patient["business_identifier"].strip()
if value:
counts[value] = counts.get(value, 0) + 1
results = []
for source_row, patient in enumerate(patients, start=2):
value = patient["business_identifier"].strip()
if not value:
continue
passed = counts[value] == 1
issue = "business_identifier is duplicated" if not passed else ""
results.append(check_result(source_row, passed, issue))
return resultsHow is uniqueness measured?
The first loop counts each populated business identifier. The second loop creates one result for every populated value so each duplicate source row is visible as a failed check.
- Save validate_data.py.
- Check that both new evaluator functions parse successfully by running:
python3 validate_data.pyWhat does this regression check show?
The baseline report should regenerate without a Python error. This proves the validity and uniqueness evaluators are syntactically sound before the dispatcher starts calling them.
Date or indentation error?
- Check that each try block lines up with its matching except ValueError block.
- Confirm that both functions end with an indented return results line.
- Still stuck? Help me debug the validity and uniqueness evaluator functions.
- Add the deceased-date sequence evaluator below check_unique_identifier using this function:
def check_deceased_sequence(patients, organizations, run_datetime, rule):
results = []
for source_row, patient in enumerate(patients, start=2):
deceased_value = patient["deceased_date"].strip()
if not deceased_value:
continue
try:
birth_date = dt.date.fromisoformat(patient["birth_date"].strip())
deceased_date = dt.date.fromisoformat(deceased_value)
passed = deceased_date >= birth_date
except ValueError:
passed = False
issue = "deceased_date is invalid or earlier than birth_date" if not passed else ""
results.append(check_result(source_row, passed, issue))
return resultsHow is chronology checked?
The evaluator checks rows with a populated death date. A passing result requires a parseable death date that falls on or after the birth date.
- Add the organization-reference evaluator directly below check_deceased_sequence using this function:
def check_organization_reference(patients, organizations, run_datetime, rule):
organization_keys = {
organization["organization_key"].strip()
for organization in organizations
if organization["organization_key"].strip()
}
results = []
for source_row, patient in enumerate(patients, start=2):
value = patient["managing_organization_key"].strip()
passed = not value or value in organization_keys
issue = "managing_organization_key does not resolve" if not passed else ""
results.append(check_result(source_row, passed, issue))
return resultsWhat is referential integrity?
Referential integrity means that a stored reference points to a record that exists. This evaluator builds a set of organization keys and checks each populated patient reference against it.
- Save validate_data.py.
- Confirm the two consistency evaluators preserve the working script by running:
python3 validate_data.pyWhat does this check confirm?
The script should complete without a Python error. This confirms the chronology and organization-reference functions can be loaded alongside the existing validator.
Seeing a set or date error?
- Check that the organization key comprehension uses curly braces around the complete expression.
- Confirm that deceased_date and birth_date match the fixture headers exactly.
- Still stuck? Help me debug the chronology and organization-reference checks.
- Add the timeliness evaluator below check_organization_reference using this function:
def check_timeliness(patients, organizations, run_datetime, rule):
maximum_age_days = int(rule["threshold_value"])
results = []
for source_row, patient in enumerate(patients, start=2):
value = patient["last_updated"].strip()
try:
parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=dt.timezone.utc)
age_days = (run_datetime - parsed.astimezone(dt.timezone.utc)).days
passed = 0 <= age_days <= maximum_age_days
except ValueError:
passed = False
issue = "last_updated is invalid or outside the allowed age" if not passed else ""
results.append(check_result(source_row, passed, issue))
return resultsHow is timeliness evaluated?
- The evaluator reads the maximum age from the rule catalog.
- It converts each populated timestamp into an aware UTC datetime.
- It passes timestamps whose age falls between zero days and the configured maximum.
- Add the checker dispatch map directly below check_timeliness using this block:
CHECKERS = {
"DQ-COMP-001": check_required_identifier,
"DQ-VAL-001": check_birth_date,
"DQ-UNIQ-001": check_unique_identifier,
"DQ-CONS-001": check_deceased_sequence,
"DQ-CONS-002": check_organization_reference,
"DQ-TIME-001": check_timeliness,
}Why use a dispatch map?
The CHECKERS dictionary connects each catalog rule to one evaluator function. This keeps rule selection explicit and makes a missing implementation fail clearly.
- Save validate_data.py.
- Confirm the timeliness function and dispatch map load successfully by running:
python3 validate_data.pyWhat does this check establish?
The script should complete without a Python error. All six rule identifiers now resolve to evaluator functions that Python can load.
Checker name not defined?
- Place CHECKERS below all six evaluator functions.
- Confirm that every function name in the dictionary matches its definition exactly.
- Still stuck? Help me fix the timeliness evaluator or CHECKERS dispatch map.
Aggregate the quality results
The evaluators now return compatible results. The final routine loads the three CSV inputs, dispatches every enabled rule, and combines the outcomes by dimension.
- Delete the existing baseline main routine while keeping the code above it.
- Build the replacement main routine by pasting the following blocks in order before saving the file:
def main():
patients, patient_fields = read_csv(PATIENTS_PATH)
organizations, organization_fields = read_csv(ORGANIZATIONS_PATH)
rules, rule_fields = read_csv(RULES_PATH)
require_columns(
patient_fields,
[
"patient_key",
"business_identifier",
"birth_date",
"deceased_date",
"active",
"managing_organization_key",
"last_updated",
],
PATIENTS_PATH,
)
require_columns(
organization_fields,
["organization_key", "organization_name"],
ORGANIZATIONS_PATH,
)What does the first part load?
The routine loads the patient, organization, and rule files with their headers. It then checks that both synthetic exports match the column contract expected by the evaluators.
- Continue the same main function with the rule contract and summary setup:
require_columns(
rule_fields,
[
"rule_id",
"dimension",
"resource_type",
"fhir_element",
"sql_field",
"description",
"severity",
"threshold_value",
"owner",
"escalation_action",
"enabled",
],
RULES_PATH,
)
enabled_rules = [
rule for rule in rules if rule["enabled"].strip().lower() == "true"
]
run_datetime = dt.datetime.now(dt.timezone.utc)
summary = {
dimension: {"total": 0, "passed": 0, "failed": 0}
for dimension in DIMENSIONS
}
quarantine_rows = []What state does this prepare?
- The rule column check protects the contract used by every evaluator.
- The enabled_rules list selects active controls from the catalog.
- The summary dictionary starts each dimension with zero totals.
- Continue the same function with rule dispatch and result counting:
for rule in enabled_rules:
checker = CHECKERS.get(rule["rule_id"])
if checker is None:
raise ValueError("No checker exists for rule_id {}".format(rule["rule_id"]))
evaluations = checker(patients, organizations, run_datetime, rule)
dimension = rule["dimension"]
if dimension not in summary:
raise ValueError("Unsupported dimension {}".format(dimension))
for evaluation in evaluations:
summary[dimension]["total"] += 1
if evaluation["passed"]:
summary[dimension]["passed"] += 1
continue
summary[dimension]["failed"] += 1How are rules dispatched?
Each enabled rule looks up its evaluator in CHECKERS. Every returned result increments the total before being counted as passed or failed.
- Continue the failed-result branch with the structured issue record:
quarantine_rows.append(
{
"quarantine_id": "Q-{:03d}".format(len(quarantine_rows) + 1),
"resource_type": rule["resource_type"],
"source_row": evaluation["source_row"],
"rule_id": rule["rule_id"],
"dimension": dimension,
"severity": rule["severity"],
"failed_field": rule["sql_field"],
"observed_issue": evaluation["issue"],
"remediation_action": rule["escalation_action"],
"owner": rule["owner"],
}
)What does a failed result retain?
Each failed check retains its source row, rule metadata, generic issue, owner, and remediation action. Direct patient keys and business identifiers are absent from this record.
- Continue the function with the dimension-level report rows:
report_rows = []
for dimension in DIMENSIONS:
counts = summary[dimension]
pass_rate = counts["passed"] / counts["total"] if counts["total"] else 1.0
report_rows.append(
{
"dimension": dimension,
"total_checks": counts["total"],
"passed_checks": counts["passed"],
"failed_checks": counts["failed"],
"pass_rate": "{:.1%}".format(pass_rate),
"status": "PASS" if counts["failed"] == 0 else "FAIL",
}
)How is each dimension summarized?
The routine calculates a pass rate from applicable checks. A dimension receives PASS only when its failed count is zero.
- Continue the function with the overall report row:
total_checks = sum(row["total_checks"] for row in report_rows)
passed_checks = sum(row["passed_checks"] for row in report_rows)
failed_checks = sum(row["failed_checks"] for row in report_rows)
overall_rate = passed_checks / total_checks if total_checks else 1.0
report_rows.append(
{
"dimension": "Overall",
"total_checks": total_checks,
"passed_checks": passed_checks,
"failed_checks": failed_checks,
"pass_rate": "{:.1%}".format(overall_rate),
"status": "PASS" if failed_checks == 0 else "FAIL",
}
)What does the overall row represent?
The overall row combines the dimension totals into one project-wide result. Its status exposes whether any applicable check failed.
- Finish the function with the output writes and terminal summary:
write_csv(REPORT_PATH, REPORT_FIELDS, report_rows)
write_csv(QUARANTINE_PATH, QUARANTINE_FIELDS, quarantine_rows)
print("Validation complete")
print("Quality report: {}".format(REPORT_PATH))
print("Quarantine: {}".format(QUARANTINE_PATH))
print("Failed checks: {}".format(failed_checks))
if __name__ == "__main__":
main()What does the final block do?
The routine writes the aggregate report and structured failure records to the output folder. The terminal summary confirms completion and prints the failed-check count.
- Save validate_data.py.
- Compare the complete file with the reference below before running it.
✔️ Awesome, I've got everything!
Your validator now contains six evaluators, a rule dispatcher, and dimension-level aggregation. Save validate_data.py before the final run.
ⓧ I'd like to double check the full code
Compare your complete validate_data.py file with this reference:
import csv
import datetime as dt
from pathlib import Path
BASE_DIR = Path.cwd()
INPUT_DIR = BASE_DIR / "input"
OUTPUT_DIR = BASE_DIR / "output"
PATIENTS_PATH = INPUT_DIR / "patients.csv"
ORGANIZATIONS_PATH = INPUT_DIR / "organizations.csv"
RULES_PATH = BASE_DIR / "quality_rules.csv"
REPORT_PATH = OUTPUT_DIR / "quality_report.csv"
QUARANTINE_PATH = OUTPUT_DIR / "quarantined_records.csv"
DIMENSIONS = [
"Completeness",
"Validity",
"Uniqueness",
"Consistency",
"Timeliness",
]
REPORT_FIELDS = [
"dimension",
"total_checks",
"passed_checks",
"failed_checks",
"pass_rate",
"status",
]
QUARANTINE_FIELDS = [
"quarantine_id",
"resource_type",
"source_row",
"rule_id",
"dimension",
"severity",
"failed_field",
"observed_issue",
"remediation_action",
"owner",
]
def read_csv(path):
with path.open("r", newline="", encoding="utf-8") as handle:
reader = csv.DictReader(handle)
rows = list(reader)
return rows, reader.fieldnames or []
def write_csv(path, fieldnames, rows):
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
def require_columns(actual_fields, required_fields, path):
missing = [field for field in required_fields if field not in actual_fields]
if missing:
raise ValueError(
"{} is missing required columns: {}".format(path, ", ".join(missing))
)
def check_result(source_row, passed, issue=""):
return {
"source_row": source_row,
"passed": passed,
"issue": issue,
}
def check_required_identifier(patients, organizations, run_datetime, rule):
results = []
for source_row, patient in enumerate(patients, start=2):
passed = bool(patient["business_identifier"].strip())
issue = "business_identifier is blank" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_birth_date(patients, organizations, run_datetime, rule):
results = []
run_date = run_datetime.date()
for source_row, patient in enumerate(patients, start=2):
value = patient["birth_date"].strip()
if not value:
results.append(check_result(source_row, True))
continue
try:
birth_date = dt.date.fromisoformat(value)
passed = birth_date <= run_date
except ValueError:
passed = False
issue = "birth_date is invalid or after the run date" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_unique_identifier(patients, organizations, run_datetime, rule):
counts = {}
for patient in patients:
value = patient["business_identifier"].strip()
if value:
counts[value] = counts.get(value, 0) + 1
results = []
for source_row, patient in enumerate(patients, start=2):
value = patient["business_identifier"].strip()
if not value:
continue
passed = counts[value] == 1
issue = "business_identifier is duplicated" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_deceased_sequence(patients, organizations, run_datetime, rule):
results = []
for source_row, patient in enumerate(patients, start=2):
deceased_value = patient["deceased_date"].strip()
if not deceased_value:
continue
try:
birth_date = dt.date.fromisoformat(patient["birth_date"].strip())
deceased_date = dt.date.fromisoformat(deceased_value)
passed = deceased_date >= birth_date
except ValueError:
passed = False
issue = "deceased_date is invalid or earlier than birth_date" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_organization_reference(patients, organizations, run_datetime, rule):
organization_keys = {
organization["organization_key"].strip()
for organization in organizations
if organization["organization_key"].strip()
}
results = []
for source_row, patient in enumerate(patients, start=2):
value = patient["managing_organization_key"].strip()
passed = not value or value in organization_keys
issue = "managing_organization_key does not resolve" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_timeliness(patients, organizations, run_datetime, rule):
maximum_age_days = int(rule["threshold_value"])
results = []
for source_row, patient in enumerate(patients, start=2):
value = patient["last_updated"].strip()
try:
parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=dt.timezone.utc)
age_days = (run_datetime - parsed.astimezone(dt.timezone.utc)).days
passed = 0 <= age_days <= maximum_age_days
except ValueError:
passed = False
issue = "last_updated is invalid or outside the allowed age" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
CHECKERS = {
"DQ-COMP-001": check_required_identifier,
"DQ-VAL-001": check_birth_date,
"DQ-UNIQ-001": check_unique_identifier,
"DQ-CONS-001": check_deceased_sequence,
"DQ-CONS-002": check_organization_reference,
"DQ-TIME-001": check_timeliness,
}
def main():
patients, patient_fields = read_csv(PATIENTS_PATH)
organizations, organization_fields = read_csv(ORGANIZATIONS_PATH)
rules, rule_fields = read_csv(RULES_PATH)
require_columns(
patient_fields,
[
"patient_key",
"business_identifier",
"birth_date",
"deceased_date",
"active",
"managing_organization_key",
"last_updated",
],
PATIENTS_PATH,
)
require_columns(
organization_fields,
["organization_key", "organization_name"],
ORGANIZATIONS_PATH,
)
require_columns(
rule_fields,
[
"rule_id",
"dimension",
"resource_type",
"fhir_element",
"sql_field",
"description",
"severity",
"threshold_value",
"owner",
"escalation_action",
"enabled",
],
RULES_PATH,
)
enabled_rules = [
rule for rule in rules if rule["enabled"].strip().lower() == "true"
]
run_datetime = dt.datetime.now(dt.timezone.utc)
summary = {
dimension: {"total": 0, "passed": 0, "failed": 0}
for dimension in DIMENSIONS
}
quarantine_rows = []
for rule in enabled_rules:
checker = CHECKERS.get(rule["rule_id"])
if checker is None:
raise ValueError("No checker exists for rule_id {}".format(rule["rule_id"]))
evaluations = checker(patients, organizations, run_datetime, rule)
dimension = rule["dimension"]
if dimension not in summary:
raise ValueError("Unsupported dimension {}".format(dimension))
for evaluation in evaluations:
summary[dimension]["total"] += 1
if evaluation["passed"]:
summary[dimension]["passed"] += 1
continue
summary[dimension]["failed"] += 1
quarantine_rows.append(
{
"quarantine_id": "Q-{:03d}".format(len(quarantine_rows) + 1),
"resource_type": rule["resource_type"],
"source_row": evaluation["source_row"],
"rule_id": rule["rule_id"],
"dimension": dimension,
"severity": rule["severity"],
"failed_field": rule["sql_field"],
"observed_issue": evaluation["issue"],
"remediation_action": rule["escalation_action"],
"owner": rule["owner"],
}
)
report_rows = []
for dimension in DIMENSIONS:
counts = summary[dimension]
pass_rate = counts["passed"] / counts["total"] if counts["total"] else 1.0
report_rows.append(
{
"dimension": dimension,
"total_checks": counts["total"],
"passed_checks": counts["passed"],
"failed_checks": counts["failed"],
"pass_rate": "{:.1%}".format(pass_rate),
"status": "PASS" if counts["failed"] == 0 else "FAIL",
}
)
total_checks = sum(row["total_checks"] for row in report_rows)
passed_checks = sum(row["passed_checks"] for row in report_rows)
failed_checks = sum(row["failed_checks"] for row in report_rows)
overall_rate = passed_checks / total_checks if total_checks else 1.0
report_rows.append(
{
"dimension": "Overall",
"total_checks": total_checks,
"passed_checks": passed_checks,
"failed_checks": failed_checks,
"pass_rate": "{:.1%}".format(overall_rate),
"status": "PASS" if failed_checks == 0 else "FAIL",
}
)
write_csv(REPORT_PATH, REPORT_FIELDS, report_rows)
write_csv(QUARANTINE_PATH, QUARANTINE_FIELDS, quarantine_rows)
print("Validation complete")
print("Quality report: {}".format(REPORT_PATH))
print("Quarantine: {}".format(QUARANTINE_PATH))
print("Failed checks: {}".format(failed_checks))
if __name__ == "__main__":
main()Before you run the expanded validator, which quality dimensions do you expect the deliberately flawed fixtures to fail?
- Run the complete validator from the integrated terminal using:
python3 validate_data.pyWhat should the terminal confirm?
You should see Validation complete followed by the output paths and a failed-check count. This confirms that all enabled rules ran and the aggregate evidence was written.
- In the Visual Studio Code Explorer sidebar, select output/quality_report.csv.
You should see Completeness, Validity, Uniqueness, Consistency, Timeliness, and Overall rows. The failing dimension statuses expose the blank identifier, future birth date, duplicate identifiers, invalid chronology, unresolved organization reference, and stale record that the baseline missed.
Report missing dimensions?
- Confirm that all six rows in quality_rules.csv end with true.
- Check that every rule identifier in the catalog appears in CHECKERS.
- Confirm that DIMENSIONS contains the five dimension names with matching capitalization.
- Still stuck? Help me diagnose why my aggregate report is missing dimensions or statuses.
That is the quality gap exposed: one plausible dataset now produces visible evidence across five dimensions. Next, you will turn each failed check into a minimized quarantine finding that a named owner can investigate safely.
Route Failures to Privacy-Safe Quarantine
Your validator now measures five quality dimensions. It also exposes every failing check in an aggregate report.
An aggregate report proves that quality is poor. It cannot direct a reviewer to the affected source row or the accountable owner.
This step adds a privacy-safe quarantine workflow. Each failure becomes a reviewable finding without copying direct patient identifiers into another file.
In this step, get ready to:
- Define the minimized quarantine schema.
- Capture one quarantine finding for every failed evaluation.
- Generate both evidence files and verify that direct patient identifiers are excluded.
Define the privacy-safe quarantine schema
A quarantine file needs enough metadata to route each defect. Data minimisation keeps the output focused on investigation details without duplicating patient keys or business identifiers.
- In validate_data.py, locate the REPORT_FIELDS list.
- Add QUARANTINE_FIELDS immediately below that list by pasting this code:
QUARANTINE_FIELDS = [
"quarantine_id",
"resource_type",
"source_row",
"rule_id",
"dimension",
"severity",
"failed_field",
"observed_issue",
"remediation_action",
"owner",
]How does this schema protect privacy?
- The quarantine_id gives each finding a tracking reference.
- The source_row points an authorized reviewer back to the original synthetic export.
- The rule metadata identifies the defect severity. It also names the remediation owner.
- The schema excludes patient_key. It also excludes business_identifier.
- Save validate_data.py.
- Confirm the schema does not interrupt the validator by running this command:
python3 validate_data.pyWhat should I see?
You should see Validation complete in the terminal. You should also see Failed checks: 7 for the supplied fixtures on 2026-10-05.
Seeing a syntax error near the field list?
- Check that every field name remains inside double quotes.
- Check that the closing square bracket appears after owner.
- Still stuck? Help me check my QUARANTINE_FIELDS list.
Capture each failed evaluation
Every failed evaluation already contains a source row plus a generic issue. The rule catalog supplies the severity plus the accountable owner.
The validator now needs a fresh collection for those combined findings. It then adds one minimized record each time an evaluation fails.
- In main(), locate the summary dictionary below the run_datetime assignment.
- Replace that dictionary with the expanded block below:
summary = {
dimension: {"total": 0, "passed": 0, "failed": 0}
for dimension in DIMENSIONS
}
quarantine_rows = []What does this collection hold?
The summary dictionary continues to hold aggregate counts. The new quarantine_rows list holds individual findings for the current run.
Starting with an empty list prevents findings from an earlier run from being carried into new evidence.
- In the failed-evaluation branch of main(), locate summary[dimension]["failed"] += 1.
- Add the quarantine record construction immediately below that line by pasting this code:
quarantine_rows.append(
{
"quarantine_id": "Q-{:03d}".format(len(quarantine_rows) + 1),
"resource_type": rule["resource_type"],
"source_row": evaluation["source_row"],
"rule_id": rule["rule_id"],
"dimension": dimension,
"severity": rule["severity"],
"failed_field": rule["sql_field"],
"observed_issue": evaluation["issue"],
"remediation_action": rule["escalation_action"],
"owner": rule["owner"],
}
)How is each finding assembled?
- The formatted quarantine_id creates sequential references such as Q-001.
- The evaluation supplies the source row plus the generic issue.
- The rule supplies the dimension plus the severity.
- The rule also supplies the failed field plus the remediation route.
- Save validate_data.py.
- Check that all failed evaluations can be collected by running this command:
python3 validate_data.pyWhat does this run prove?
You should see Failed checks: 7 again. This confirms that all seven failing branches complete while building the in-memory quarantine records.
Seeing an indentation or name error?
- Keep quarantine_rows = [] aligned with the summary assignment inside main().
- Keep quarantine_rows.append( aligned with the failed-count increment inside the evaluation loop.
- Still stuck? Help me fix my quarantine row construction.
Write and verify the quarantine file
The findings currently exist during the validator run. Writing them through the existing CSV helper turns them into reviewable evidence beside the aggregate report.
- Near the end of main(), locate the call that writes REPORT_PATH.
- Replace that call with the two output calls below:
write_csv(REPORT_PATH, REPORT_FIELDS, report_rows)
write_csv(QUARANTINE_PATH, QUARANTINE_FIELDS, quarantine_rows)What do these output calls create?
The first call preserves the aggregate report. The second call writes one row from quarantine_rows for every failed evaluation.
Both files use explicit field lists. That boundary prevents extra source columns from leaking into the generated evidence.
- Save validate_data.py.
Before you run the completed validator, do you expect the number of quarantine findings to match the number of failed checks?
- Generate both evidence files by running this command:
python3 validate_data.pyWhat should I see?
You should see Validation complete followed by paths for both generated files. The final line should read Failed checks: 7 on 2026-10-05.
The matching count proves that every failed check produced one quarantine finding.
- In the Explorer sidebar from earlier, select output/quality_report.csv.
The Overall row should show 20 total checks. It should also show 13 passed checks plus 7 failed checks with a FAIL status.
- In the Explorer sidebar, select output/quarantined_records.csv.
You should see seven findings under quarantine_id,resource_type,source_row,rule_id,dimension,severity,failed_field,observed_issue,remediation_action,owner.
You should not see a patient_key column. You should not see a business_identifier column.
Missing the quarantine file or seven findings?
- Check that the second write_csv call remains inside main().
- Check that quarantine_rows.append( remains inside the failed-evaluation branch.
- Still stuck? Help me trace my missing quarantine output.
✔️ Awesome, I've got everything!
Great work. Save validate_data.py after confirming that both output files contain the expected evidence.
ⓧ I'd like to double check the full code
Compare your complete validate_data.py file with this reference:
import csv
import datetime as dt
from pathlib import Path
BASE_DIR = Path.cwd()
INPUT_DIR = BASE_DIR / "input"
OUTPUT_DIR = BASE_DIR / "output"
PATIENTS_PATH = INPUT_DIR / "patients.csv"
ORGANIZATIONS_PATH = INPUT_DIR / "organizations.csv"
RULES_PATH = BASE_DIR / "quality_rules.csv"
REPORT_PATH = OUTPUT_DIR / "quality_report.csv"
QUARANTINE_PATH = OUTPUT_DIR / "quarantined_records.csv"
DIMENSIONS = [
"Completeness",
"Validity",
"Uniqueness",
"Consistency",
"Timeliness",
]
REPORT_FIELDS = [
"dimension",
"total_checks",
"passed_checks",
"failed_checks",
"pass_rate",
"status",
]
QUARANTINE_FIELDS = [
"quarantine_id",
"resource_type",
"source_row",
"rule_id",
"dimension",
"severity",
"failed_field",
"observed_issue",
"remediation_action",
"owner",
]
def read_csv(path):
with path.open("r", newline="", encoding="utf-8") as handle:
reader = csv.DictReader(handle)
rows = list(reader)
return rows, reader.fieldnames or []
def write_csv(path, fieldnames, rows):
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
def require_columns(actual_fields, required_fields, path):
missing = [field for field in required_fields if field not in actual_fields]
if missing:
raise ValueError(
"{} is missing required columns: {}".format(path, ", ".join(missing))
)
def check_result(source_row, passed, issue=""):
return {
"source_row": source_row,
"passed": passed,
"issue": issue,
}
def check_required_identifier(patients, organizations, run_datetime, rule):
results = []
for source_row, patient in enumerate(patients, start=2):
passed = bool(patient["business_identifier"].strip())
issue = "business_identifier is blank" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_birth_date(patients, organizations, run_datetime, rule):
results = []
run_date = run_datetime.date()
for source_row, patient in enumerate(patients, start=2):
value = patient["birth_date"].strip()
if not value:
results.append(check_result(source_row, True))
continue
try:
birth_date = dt.date.fromisoformat(value)
passed = birth_date <= run_date
except ValueError:
passed = False
issue = "birth_date is invalid or after the run date" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_unique_identifier(patients, organizations, run_datetime, rule):
counts = {}
for patient in patients:
value = patient["business_identifier"].strip()
if value:
counts[value] = counts.get(value, 0) + 1
results = []
for source_row, patient in enumerate(patients, start=2):
value = patient["business_identifier"].strip()
if not value:
continue
passed = counts[value] == 1
issue = "business_identifier is duplicated" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_deceased_sequence(patients, organizations, run_datetime, rule):
results = []
for source_row, patient in enumerate(patients, start=2):
deceased_value = patient["deceased_date"].strip()
if not deceased_value:
continue
try:
birth_date = dt.date.fromisoformat(patient["birth_date"].strip())
deceased_date = dt.date.fromisoformat(deceased_value)
passed = deceased_date >= birth_date
except ValueError:
passed = False
issue = "deceased_date is invalid or earlier than birth_date" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_organization_reference(patients, organizations, run_datetime, rule):
organization_keys = {
organization["organization_key"].strip()
for organization in organizations
if organization["organization_key"].strip()
}
results = []
for source_row, patient in enumerate(patients, start=2):
value = patient["managing_organization_key"].strip()
passed = not value or value in organization_keys
issue = "managing_organization_key does not resolve" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
def check_timeliness(patients, organizations, run_datetime, rule):
maximum_age_days = int(rule["threshold_value"])
results = []
for source_row, patient in enumerate(patients, start=2):
value = patient["last_updated"].strip()
try:
parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=dt.timezone.utc)
age_days = (run_datetime - parsed.astimezone(dt.timezone.utc)).days
passed = 0 <= age_days <= maximum_age_days
except ValueError:
passed = False
issue = "last_updated is invalid or outside the allowed age" if not passed else ""
results.append(check_result(source_row, passed, issue))
return results
CHECKERS = {
"DQ-COMP-001": check_required_identifier,
"DQ-VAL-001": check_birth_date,
"DQ-UNIQ-001": check_unique_identifier,
"DQ-CONS-001": check_deceased_sequence,
"DQ-CONS-002": check_organization_reference,
"DQ-TIME-001": check_timeliness,
}
def main():
patients, patient_fields = read_csv(PATIENTS_PATH)
organizations, organization_fields = read_csv(ORGANIZATIONS_PATH)
rules, rule_fields = read_csv(RULES_PATH)
require_columns(
patient_fields,
[
"patient_key",
"business_identifier",
"birth_date",
"deceased_date",
"active",
"managing_organization_key",
"last_updated",
],
PATIENTS_PATH,
)
require_columns(
organization_fields,
["organization_key", "organization_name"],
ORGANIZATIONS_PATH,
)
require_columns(
rule_fields,
[
"rule_id",
"dimension",
"resource_type",
"fhir_element",
"sql_field",
"description",
"severity",
"threshold_value",
"owner",
"escalation_action",
"enabled",
],
RULES_PATH,
)
enabled_rules = [
rule for rule in rules if rule["enabled"].strip().lower() == "true"
]
run_datetime = dt.datetime.now(dt.timezone.utc)
summary = {
dimension: {"total": 0, "passed": 0, "failed": 0}
for dimension in DIMENSIONS
}
quarantine_rows = []
for rule in enabled_rules:
checker = CHECKERS.get(rule["rule_id"])
if checker is None:
raise ValueError("No checker exists for rule_id {}".format(rule["rule_id"]))
evaluations = checker(patients, organizations, run_datetime, rule)
dimension = rule["dimension"]
if dimension not in summary:
raise ValueError("Unsupported dimension {}".format(dimension))
for evaluation in evaluations:
summary[dimension]["total"] += 1
if evaluation["passed"]:
summary[dimension]["passed"] += 1
continue
summary[dimension]["failed"] += 1
quarantine_rows.append(
{
"quarantine_id": "Q-{:03d}".format(len(quarantine_rows) + 1),
"resource_type": rule["resource_type"],
"source_row": evaluation["source_row"],
"rule_id": rule["rule_id"],
"dimension": dimension,
"severity": rule["severity"],
"failed_field": rule["sql_field"],
"observed_issue": evaluation["issue"],
"remediation_action": rule["escalation_action"],
"owner": rule["owner"],
}
)
report_rows = []
for dimension in DIMENSIONS:
counts = summary[dimension]
pass_rate = counts["passed"] / counts["total"] if counts["total"] else 1.0
report_rows.append(
{
"dimension": dimension,
"total_checks": counts["total"],
"passed_checks": counts["passed"],
"failed_checks": counts["failed"],
"pass_rate": "{:.1%}".format(pass_rate),
"status": "PASS" if counts["failed"] == 0 else "FAIL",
}
)
total_checks = sum(row["total_checks"] for row in report_rows)
passed_checks = sum(row["passed_checks"] for row in report_rows)
failed_checks = sum(row["failed_checks"] for row in report_rows)
overall_rate = passed_checks / total_checks if total_checks else 1.0
report_rows.append(
{
"dimension": "Overall",
"total_checks": total_checks,
"passed_checks": passed_checks,
"failed_checks": failed_checks,
"pass_rate": "{:.1%}".format(overall_rate),
"status": "PASS" if failed_checks == 0 else "FAIL",
}
)
write_csv(REPORT_PATH, REPORT_FIELDS, report_rows)
write_csv(QUARANTINE_PATH, QUARANTINE_FIELDS, quarantine_rows)
print("Validation complete")
print("Quality report: {}".format(REPORT_PATH))
print("Quarantine: {}".format(QUARANTINE_PATH))
print("Failed checks: {}".format(failed_checks))
if __name__ == "__main__":
main()What should you compare?
Check the QUARANTINE_FIELDS order first. Then compare the failed-evaluation branch plus the two final write_csv calls.
You have turned seven anonymous failure counts into seven actionable findings. Each finding now carries enough context for remediation without exposing direct patient identifiers.
Next, you will turn these controls plus their generated evidence into a governance brief and portfolio case study.
Turn Checks into Portfolio Evidence
Your validator now turns six rules into an aggregate report plus a privacy-safe quarantine file. Those outputs provide the evidence that a reviewer needs to assess the data product.
This step connects that evidence to ownership, escalation, privacy controls, and release decisions. You will package the result as a Markdown case study that a reviewer can inspect in Visual Studio Code.
In this step, get ready to:
- Document governance ownership, escalation, privacy controls, and release rules.
- Capture non-identifying evidence from the generated outputs.
- Build a portfolio README that connects the problem, implementation, evidence, and limitations.
Document governance decisions
Governance assigns people and decisions to technical findings. The brief defines who handles each severity level, who can access row-level evidence, and what must happen before release.
- Switch back to the empty governance_brief.md file from earlier.
- Add the purpose, scope, privacy principles, and guidance links by pasting this first section:
# Governance Brief: FHIR-to-SQL Quality Layer
## Purpose
This control layer checks whether a synthetic Patient export is fit for downstream analytical use. It reports defects, routes remediation, and preserves the source instead of silently correcting clinical or administrative facts.
## Scope and classification
- The supplied fixtures are synthetic and must not be replaced with real patient data for a public portfolio demonstration.
- `quality_report.csv` contains aggregate counts only.
- `quarantined_records.csv` contains source row numbers and operational issue metadata. Treat it as sensitive because row-level metadata may become identifying when combined with source data.
- Direct patient keys and business identifiers are intentionally excluded from both outputs.
- A production implementation would require the organisation's lawful basis, privacy review, access controls, retention schedule, and incident procedures.
## Principles applied
- Data minimisation: outputs contain only the evidence needed to investigate a rule failure.
- Accuracy: detected defects are routed for verification rather than automatically overwritten.
- Storage limitation: quarantine evidence is temporary and tied to remediation.
- Integrity and confidentiality: access is limited to authorised roles and outputs are not published with real data.
- Accountability: every rule names an owner and an escalation action.
Relevant guidance:
- ICO data protection principles: https://ico.org.uk/for-organisations/uk-gdpr-guidance-and-resources/data-protection-principles/a-guide-to-the-data-protection-principles/
- ICO pseudonymisation guidance: https://ico.org.uk/for-organisations/uk-gdpr-guidance-and-resources/data-sharing/anonymisation/pseudonymisation/How does this section protect the evidence?
- The scope keeps public evidence tied to synthetic fixtures.
- Data minimisation limits each output to information needed for review or remediation.
- Accuracy preserves the source while an accountable owner investigates each defect.
- Storage limitation gives quarantine evidence a temporary purpose.
- Save governance_brief.md.
- Press Cmd+K V to open the Markdown preview to the side.
You should see the purpose, classification, privacy principles, and two guidance links rendered as a structured brief.
Is the Markdown preview blank?
- Confirm that the content was pasted into governance_brief.md.
- Check that the file is saved with the .md extension.
- Still stuck? Help me troubleshoot the governance brief preview.
The rule catalog already names owners and severity levels. The next section turns those values into explicit accountability and handling decisions.
- Return to the bottom of governance_brief.md.
- Add the ownership and escalation matrices by pasting this section:
## Ownership matrix
| Role | Accountability |
| --- | --- |
| Data Quality Owner | Owns the rule catalog, approves thresholds, reviews trends, and makes the final data release recommendation. |
| FHIR Data Steward | Investigates FHIR semantics, identifier defects, and source-to-SQL mapping issues. |
| Clinical Data Steward | Reviews chronology or semantic defects that could affect clinical meaning. |
| Pipeline Engineer | Repairs transformations and reference mappings, then reruns validation. |
| Privacy Lead | Approves access, public evidence, retention, and any use of non-synthetic data. |
## Escalation matrix
| Severity | Handling decision |
| --- | --- |
| Critical | Block release immediately. Notify the Clinical Data Steward and Data Quality Owner. Require documented correction and a clean rerun. |
| High | Block release of affected rows. Assign the named owner and require correction or an approved exception before release. |
| Medium | Record the issue, confirm the refresh expectation, and resolve it before the next approved publication. |How do the matrices work together?
The ownership matrix defines each role's accountability. The escalation matrix converts rule severity into a concrete handling decision.
Together, they give every failed check a route from detection to review.
- Save governance_brief.md.
- Check the live preview for two rendered tables with five ownership roles and three severity levels.
You should see separate ownership and escalation tables beneath the privacy principles.
Are the tables showing as plain text?
- Check that every table row begins and ends with a vertical bar.
- Confirm that each header has a separator row containing hyphens.
- Still stuck? Help me fix the Markdown tables.
The final part of the brief defines the lifecycle of a finding. It preserves the source, restricts access, records remediation, and sets the release threshold.
- Return to the bottom of governance_brief.md.
- Complete the brief by pasting the quarantine workflow, release rule, and limitations:
## Quarantine workflow
1. Keep the source export unchanged.
2. Write one minimized issue record per failed check.
3. Restrict access to the Data Quality Owner, assigned steward, Pipeline Engineer, and Privacy Lead.
4. Investigate the source and transformation before changing any value.
5. Delete quarantine evidence after remediation approval or within 30 days, whichever comes first.
6. Rerun the validator and retain only the approved aggregate evidence for the portfolio.
## Release rule
The default decision is BLOCKED when any Critical or High finding remains unresolved. Medium findings require a documented owner, due condition, and approved exception if publication proceeds.
## Limitations
This is a local portfolio control, not a clinical safety system, FHIR conformance validator, legal compliance assessment, or production access-control implementation. The 365-day timeliness threshold and escalation periods are demonstration policy choices that an accountable organisation must approve for its own purpose.What makes this a governance workflow?
- The workflow keeps the original export unchanged while an owner investigates the defect.
- The access rule limits row-level evidence to named operational roles.
- The retention rule removes quarantine evidence after approval or within the documented period.
- The release rule blocks unresolved Critical or High findings.
- Save governance_brief.md.
- Check the preview for a six-stage workflow, a blocked release rule, and a limitations section.
Your brief now connects privacy, ownership, remediation, retention, and release decisions to the validator's findings.
Is part of the brief missing?
- Confirm that the quarantine workflow appears after the escalation matrix.
- Check that the release rule contains the Critical and High severity conditions.
- Still stuck? Help me compare my governance brief.
✔️ Awesome, I've got everything!
Your governance brief now documents the complete control and decision process.
ⓧ I'd like to double check the full code
- Compare governance_brief.md with this complete reference:
# Governance Brief: FHIR-to-SQL Quality Layer
## Purpose
This control layer checks whether a synthetic Patient export is fit for downstream analytical use. It reports defects, routes remediation, and preserves the source instead of silently correcting clinical or administrative facts.
## Scope and classification
- The supplied fixtures are synthetic and must not be replaced with real patient data for a public portfolio demonstration.
- `quality_report.csv` contains aggregate counts only.
- `quarantined_records.csv` contains source row numbers and operational issue metadata. Treat it as sensitive because row-level metadata may become identifying when combined with source data.
- Direct patient keys and business identifiers are intentionally excluded from both outputs.
- A production implementation would require the organisation's lawful basis, privacy review, access controls, retention schedule, and incident procedures.
## Principles applied
- Data minimisation: outputs contain only the evidence needed to investigate a rule failure.
- Accuracy: detected defects are routed for verification rather than automatically overwritten.
- Storage limitation: quarantine evidence is temporary and tied to remediation.
- Integrity and confidentiality: access is limited to authorised roles and outputs are not published with real data.
- Accountability: every rule names an owner and an escalation action.
Relevant guidance:
- ICO data protection principles: https://ico.org.uk/for-organisations/uk-gdpr-guidance-and-resources/data-protection-principles/a-guide-to-the-data-protection-principles/
- ICO pseudonymisation guidance: https://ico.org.uk/for-organisations/uk-gdpr-guidance-and-resources/data-sharing/anonymisation/pseudonymisation/
## Ownership matrix
| Role | Accountability |
| --- | --- |
| Data Quality Owner | Owns the rule catalog, approves thresholds, reviews trends, and makes the final data release recommendation. |
| FHIR Data Steward | Investigates FHIR semantics, identifier defects, and source-to-SQL mapping issues. |
| Clinical Data Steward | Reviews chronology or semantic defects that could affect clinical meaning. |
| Pipeline Engineer | Repairs transformations and reference mappings, then reruns validation. |
| Privacy Lead | Approves access, public evidence, retention, and any use of non-synthetic data. |
## Escalation matrix
| Severity | Handling decision |
| --- | --- |
| Critical | Block release immediately. Notify the Clinical Data Steward and Data Quality Owner. Require documented correction and a clean rerun. |
| High | Block release of affected rows. Assign the named owner and require correction or an approved exception before release. |
| Medium | Record the issue, confirm the refresh expectation, and resolve it before the next approved publication. |
## Quarantine workflow
1. Keep the source export unchanged.
2. Write one minimized issue record per failed check.
3. Restrict access to the Data Quality Owner, assigned steward, Pipeline Engineer, and Privacy Lead.
4. Investigate the source and transformation before changing any value.
5. Delete quarantine evidence after remediation approval or within 30 days, whichever comes first.
6. Rerun the validator and retain only the approved aggregate evidence for the portfolio.
## Release rule
The default decision is BLOCKED when any Critical or High finding remains unresolved. Medium findings require a documented owner, due condition, and approved exception if publication proceeds.
## Limitations
This is a local portfolio control, not a clinical safety system, FHIR conformance validator, legal compliance assessment, or production access-control implementation. The 365-day timeliness threshold and escalation periods are demonstration policy choices that an accountable organisation must approve for its own purpose.Capture privacy-safe evidence
The generated report and quarantine file show the quality outcome plus the route for each finding. A carefully framed screenshot lets you demonstrate both outputs without exposing the synthetic source identifiers.
Keep the capture non-identifying
Capturing healthcare evidence can feel risky. These generated outputs contain aggregate results and minimized operational findings, so keep the input fixtures outside the selected screenshot area.
Confirm that the quarantine headers exclude patient_key and business_identifier before you capture anything.
- Select output/quality_report.csv in the Visual Studio Code Explorer sidebar.
- Select output/quarantined_records.csv in the Explorer sidebar.
- Drag the output/quarantined_records.csv editor tab to the right edge of the editor.
- Check the quarantine header for the ten approved operational fields.
- Confirm that no direct patient identifier column is visible.
You should now see the aggregate quality evidence beside the minimized quarantine evidence. The source fixture remains outside the view.
- Press Shift+Command+4 to start a selected-area screenshot.
- Drag the crosshair around only the two generated output panes.
- Release the pointer to save the screenshot to your Desktop.
Finding the newest screenshot among Desktop files can be fiddly. Its timestamp identifies the capture you just made.
- Open your Desktop in Finder.
- Rename the newest screenshot to data-quality-results.png.
- Drag data-quality-results.png into the docs folder in the Visual Studio Code Explorer sidebar.
- Confirm that docs/data-quality-results.png appears in the Explorer sidebar.
That evidence is now ready for the case study. It shows the failed checks and remediation metadata without exposing direct patient keys or business identifiers.
Can't find the screenshot?
- Check the Desktop for the most recently created image file.
- Confirm that the renamed file ends with .png.
- Still stuck? Help me place the screenshot in the docs folder.
Build the portfolio case study
A case study gives the evidence a clear narrative. It explains the quality problem, shows how the validator works, links each artifact, and states the boundaries of the demonstration.
- Switch back to the empty README.md file from earlier.
- Add the project summary, problem, and architecture by pasting this first section:
# FHIR-to-SQL Data Governance and Quality Layer
A local portfolio extension that turns a transformed healthcare dataset into a governed data product with executable quality rules, aggregate evidence, privacy-safe quarantine metadata, ownership, and escalation decisions.
## What problem this solves
A successful FHIR-to-SQL transformation can still contain missing identifiers, duplicate business identifiers, invalid dates, inconsistent chronology, broken references, or stale records. This project detects those issues before the data is presented as trustworthy.
## Architecture
```text
Synthetic Patient and Organization CSV exports
|
v
quality_rules.csv
|
v
validate_data.py
/ \
v v
quality_report.csv quarantined_records.csv
\ /
v v
governance brief and portfolio evidence
```What does the architecture show?
The diagram traces synthetic exports through the rule catalog and validator. It ends with two generated evidence files plus the governance documentation.
This gives a reviewer the complete flow without requiring access to the source code first.
- Save README.md.
- Press Cmd+K V to open the README preview to the side.
You should see a project heading, a problem statement, and a text architecture diagram with two output branches.
Is the architecture misaligned?
- Confirm that the architecture remains inside a fenced text block.
- Check that the backslashes beside the output files were preserved.
- Still stuck? Help me fix the README architecture diagram.
The next section connects each quality dimension to an observable control. It also gives a reviewer one command and a reproducible demonstration result.
- Return to the bottom of README.md.
- Add the quality dimensions, run command, and expected result by pasting this section:
## Quality dimensions
| Dimension | Example control |
| --- | --- |
| Completeness | Business identifier is populated. |
| Validity | Birth date is parseable and not in the future. |
| Uniqueness | Populated business identifiers are unique. |
| Consistency | Death chronology is valid and organization references resolve. |
| Timeliness | Last-updated instant is no more than 365 days old. |
## Run locally
From the `governance-quality-layer` folder:
```bash
python3 validate_data.py
```
The script creates:
- `output/quality_report.csv`
- `output/quarantined_records.csv`
## Expected demonstration result
With the supplied fixtures on 2026-10-05, the validator evaluates 20 applicable checks, passes 13, fails 7, and writes 7 minimized quarantine findings. Dates are compared with the actual UTC run time, so timeliness results may change when the project is run later.Why include expected results?
The expected result gives a reviewer a baseline for the supplied fixtures. The date caveat explains why later timeliness results can change while the validator remains correct.
The five-row table also shows that the catalog covers more than one type of defect.
- Save README.md.
- Check the live preview for five quality dimensions, one run command, two output paths, and the 20-check demonstration result.
The README now tells a reviewer how to reproduce the evidence and what the supplied fixtures demonstrate.
Is the command missing from the preview?
- Confirm that the run command sits between matching fenced code markers.
- Check that the two output paths appear immediately below the command section.
- Still stuck? Help me fix the README run section.
The case study also needs to show how privacy was designed into the outputs. The evidence section links the generated files, rule catalog, governance brief, and screenshot.
- Return to the bottom of README.md.
- Add the privacy design and evidence links by pasting this section:
## Privacy design
- The fixtures are synthetic.
- The report contains aggregate counts only.
- The quarantine output excludes `patient_key` and `business_identifier`.
- Source rows are not modified automatically.
- Row-level quarantine evidence is still treated as sensitive because it can be linked back to a source export.
See [governance_brief.md](governance_brief.md) for ownership, escalation, access, retention, and remediation decisions.
## Evidence
- [Quality report](output/quality_report.csv)
- [Privacy-safe quarantine](output/quarantined_records.csv)
- [Rule catalog](quality_rules.csv)
- [Governance brief](governance_brief.md)
What does this evidence prove?
The privacy section states that the fixtures are synthetic and the quarantine output excludes direct identifiers. It also acknowledges that row-level operational metadata still needs controlled handling.
The links let a reviewer move from the summary to the generated evidence, executable rules, governance decisions, and screenshot.
- Save README.md.
- Check the live preview for four evidence links and the screenshot from docs/data-quality-results.png.
You should see the non-identifying screenshot beneath the evidence links.
Is the screenshot missing?
- Confirm that the image exists at docs/data-quality-results.png.
- Check that the filename uses the same lowercase spelling as the README image link.
- Still stuck? Help me fix the README screenshot link.
The final sections summarize the practical skills demonstrated by the project. They also set clear limits around clinical safety, FHIR conformance, real patient data, and production access controls.
- Return to the bottom of README.md.
- Complete the case study by pasting the skills and limitations sections:
## Skills demonstrated
- Translating FHIR-aligned fields into executable SQL-export controls
- Designing reusable rules across five data quality dimensions
- Detecting missing values, invalid dates, duplicates, chronology defects, orphan references, and stale records
- Producing aggregate quality evidence and minimized quarantine records
- Assigning owners, severity, escalation, retention, and release decisions
## Limitations
This project is a portfolio demonstration. It does not validate complete FHIR resources, replace organisation-specific governance, process real patient data, or provide production authentication and authorisation controls.Why state the limitations clearly?
The skills section shows what you implemented and what the evidence demonstrates. The limitations prevent a portfolio control from being presented as a production clinical or compliance system.
Before you inspect the finished preview, which sections do you expect a reviewer to use when checking reproducibility, privacy, and governance?
- Save README.md.
- Inspect the live preview from the title through the limitations section.
- Confirm that the case study shows the architecture, five dimensions, run command, expected result, privacy design, evidence links, screenshot, skills, and limitations.
You should see the complete case study with the embedded non-identifying screenshot. The expected demonstration result should show 20 checks, 13 passes, 7 failures, and 7 minimized quarantine findings.
Does the README stop early?
- Confirm that every section was pasted below the previous section in README.md.
- Check that every fenced code block has an opening marker and a closing marker.
- Still stuck? Help me compare my complete README.
✔️ Awesome, I've got everything!
Your README now presents the quality layer as a reproducible case study with linked evidence and explicit limitations.
ⓧ I'd like to double check the full code
- Compare README.md with this complete reference:
# FHIR-to-SQL Data Governance and Quality Layer
A local portfolio extension that turns a transformed healthcare dataset into a governed data product with executable quality rules, aggregate evidence, privacy-safe quarantine metadata, ownership, and escalation decisions.
## What problem this solves
A successful FHIR-to-SQL transformation can still contain missing identifiers, duplicate business identifiers, invalid dates, inconsistent chronology, broken references, or stale records. This project detects those issues before the data is presented as trustworthy.
## Architecture
```text
Synthetic Patient and Organization CSV exports
|
v
quality_rules.csv
|
v
validate_data.py
/ \
v v
quality_report.csv quarantined_records.csv
\ /
v v
governance brief and portfolio evidence
```
## Quality dimensions
| Dimension | Example control |
| --- | --- |
| Completeness | Business identifier is populated. |
| Validity | Birth date is parseable and not in the future. |
| Uniqueness | Populated business identifiers are unique. |
| Consistency | Death chronology is valid and organization references resolve. |
| Timeliness | Last-updated instant is no more than 365 days old. |
## Run locally
From the `governance-quality-layer` folder:
```bash
python3 validate_data.py
```
The script creates:
- `output/quality_report.csv`
- `output/quarantined_records.csv`
## Expected demonstration result
With the supplied fixtures on 2026-10-05, the validator evaluates 20 applicable checks, passes 13, fails 7, and writes 7 minimized quarantine findings. Dates are compared with the actual UTC run time, so timeliness results may change when the project is run later.
## Privacy design
- The fixtures are synthetic.
- The report contains aggregate counts only.
- The quarantine output excludes `patient_key` and `business_identifier`.
- Source rows are not modified automatically.
- Row-level quarantine evidence is still treated as sensitive because it can be linked back to a source export.
See [governance_brief.md](governance_brief.md) for ownership, escalation, access, retention, and remediation decisions.
## Evidence
- [Quality report](output/quality_report.csv)
- [Privacy-safe quarantine](output/quarantined_records.csv)
- [Rule catalog](quality_rules.csv)
- [Governance brief](governance_brief.md)
## Skills demonstrated
- Translating FHIR-aligned fields into executable SQL-export controls
- Designing reusable rules across five data quality dimensions
- Detecting missing values, invalid dates, duplicates, chronology defects, orphan references, and stale records
- Producing aggregate quality evidence and minimized quarantine records
- Assigning owners, severity, escalation, retention, and release decisions
## Limitations
This project is a portfolio demonstration. It does not validate complete FHIR resources, replace organisation-specific governance, process real patient data, or provide production authentication and authorisation controls.You have turned executable checks into reviewable governance evidence. The finished project now shows what failed, who owns each finding, how privacy is protected, what blocks release, and where the evidence can be inspected.
Secret mission
Run a Governance Tabletop Exercise
The validator has exposed defects across all five quality dimensions. Use the report and privacy-safe quarantine evidence to make a release decision, assign remediation, and define what must happen before approval.
Clean Up Your Resources
Clean Up Your Resources
The local governance-quality-layer folder creates no ongoing cloud or service costs. Choose the cleanup option that fits your next move.
Resources you used:
- The local governance-quality-layer folder, including every artifact inside it.
Keep everything running
Keeping everything requires no cleanup. Choose this option if you want to refine the governance controls or present the case study again.
- The six-rule catalog remains available for future policy changes.
- The complete validator remains ready for another local run.
- The generated report and privacy-safe quarantine remain available as evidence.
- The screenshot, README, governance brief, and Governance Decision Record remain ready for review.
Pause - I'll come back to this later
Pausing ends the editing session without deleting the workspace. Every project artifact stays available for your return.
- Confirm that the validator rerun has finished in the integrated terminal.
- Close the Visual Studio Code window from earlier.
- Leave the governance-quality-layer folder on your Mac.
The validator exits after writing its files, so pausing does not require a separate stop command.
Delete - I don't want to use this again
Deletion is for the point when you no longer need the local case study. This option removes the entire workspace.
Deletion is permanent
Deleting the folder permanently removes every project artifact inside it.
The command leaves your pre-existing Python and Visual Studio Code installations in place.
- Switch back to the integrated terminal from earlier.
- Move outside the governance-quality-layer folder by running:
cd ..Why move first?
This moves the terminal into the folder that contains your workspace. The shell can now remove the workspace as one complete folder.
- Delete the governance-quality-layer folder by running:
rm -rf governance-quality-layerWhat does this command remove?
This removes the workspace and every nested artifact in one cleanup action. The synthetic fixtures, generated outputs, screenshot, README, governance brief, and decision record are deleted together.
Before you verify, do you expect the workspace name to remain in its parent folder?
- List the contents of the parent folder by running:
lsWhat should you see?
You should see the contents of the parent folder. The governance-quality-layer name should be absent.
Still see the workspace?
- Check that your terminal prompt is outside governance-quality-layer.
- Repeat the deletion command from the folder that contains governance-quality-layer.
Help me diagnose why the local project folder is still present after deletion.
That completes the cleanup. Your Mac no longer holds the project's synthetic fixtures or generated evidence.
Nice Work!
Nice Work!
Outstanding work! Your FHIR-to-SQL governance layer now turns deliberately flawed synthetic exports into reviewable quality evidence. It routes each defect while keeping direct patient identifiers out of the generated outputs.
You've learned how to:
- Create a reusable CSV rule catalog that maps FHIR elements to SQL-export fields across five data quality dimensions.
- Implement a dependency-free Python validator that runs six executable rules against synthetic Patient data. The supplied demonstration records 20 applicable checks with 7 failures. Those failures feed a privacy-safe quarantine workflow.
- Apply data minimisation by excluding patient keys and business identifiers from generated evidence. Document accountable ownership in a governance brief. Present the result through a screenshot-backed case study that uses synthetic data only.
- Complete the Secret Mission by using the quality evidence to defend a BLOCKED release decision. Assign remediation to the named stewards plus the Pipeline Engineer. Require a clean rerun before approval.
Ready to quiz yourself?