Skip to content
This repository was archived by the owner on Jul 20, 2026. It is now read-only.

fix: scope schema-drift check to litigation cases #84

Merged
damo-da merged 2 commits into from
Jun 23, 2026
Merged

fix: scope schema-drift check to litigation cases #84

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7efbe55
fix(scrapers): scope schema-drift check to litigation cases
damo-da Jun 23, 2026
f661c72
fix(scrapers): apply black, scope drift check to recent rows, validat…
damo-da Jun 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 47 additions & 13 deletions scripts/check_schema_drift.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,36 +1,70 @@
import os
from sqlalchemy import text
import sys
from sqlalchemy import text
from ngm.database.models import get_engine

# A blank case_type only signals selector/schema drift for *litigation* cases.
# Non-litigation registrations (marriage, power-of-attorney, guardianship,
# adoption, death-declaration) are filed under a numeric case-number namespace
# (e.g. 082-02-0001) and legitimately carry no case_type. Litigation cases use
# an alphabetic namespace (CP, FN, CR, WO, C1, ...). We therefore only count
# blanks among alpha-coded rows, and fail on the *ratio* rather than an absolute
# count so the check scales with table size.
LITIGATION_ONLY = "split_part(case_number, '-', 2) ~ '[A-Za-z]'"

# Fail if more than this fraction of in-scope litigation rows lack a case_type.
DRIFT_RATIO_THRESHOLD = 0.02

def check():

def check(courts=None):
db_url = os.getenv("DATABASE_URL")
if not db_url:
print("DATABASE_URL not set")
sys.exit(1)
Comment on lines +46 to 50

@gemini-code-assist gemini-code-assist Bot Jun 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

⚠️ Medium: Silent Skip on Invalid Court Identifiers

If a user passes an invalid court identifier (e.g., due to a typo like supremee), the script currently queries court_identifier = ANY(['supremee']). Since no rows match, total will be 0, and the script will print:
Schema check skipped: no litigation rows in scope (courts=['supremee'])
and exit with code 0 (success).

This silently skips the schema check instead of alerting the user that they provided an invalid court name.

Recommendation

Validate the provided court identifiers using the existing is_valid_court_identifier utility before running the query.

Suggested change
def check(courts=None):
db_url = os.getenv("DATABASE_URL")
if not db_url:
print("DATABASE_URL not set")
sys.exit(1)
def check(courts=None):
db_url = os.getenv("DATABASE_URL")
if not db_url:
print("DATABASE_URL not set")
sys.exit(1)
if courts:
from ngm.utils.court_mapping import is_valid_court_identifier
invalid = [c for c in courts if not is_valid_court_identifier(c)]
if invalid:
print(f"ERROR: Invalid court identifier(s): {invalid}")
sys.exit(1)


where = [LITIGATION_ONLY]
params = {}
if courts:
where.append("court_identifier = ANY(:courts)")
params["courts"] = list(courts)
scope = " AND ".join(where)

@gemini-code-assist gemini-code-assist Bot Jun 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

⚠️ High: Drift Ratio Dilution & Performance Bottleneck

Calculating the schema drift ratio over the entire history of the court_cases table (1.5M+ rows) introduces two major issues:

  1. Dilution of Drift (False Pass): If the scraper starts failing today (e.g., 100% of today's 1,000 new cases are missing case_type), the ratio of missing cases over the entire 1.5M historical rows will be 1,000 / 1,501,000 ≈ 0.06%. This is far below the 2% threshold (DRIFT_RATIO_THRESHOLD), meaning the check will pass and the drift will go undetected.
  2. Performance: Evaluating the regex split_part(case_number, '-', 2) ~ '[A-Za-z]' via a full table scan on 1.5M+ rows on every cron job run is extremely slow and does not scale.

Recommendation

Scope the check to recently created cases (e.g., the last 3 days). This ensures that a sudden scraper failure on a new batch will immediately spike the ratio to ~100% and trigger the alert, while also keeping the query highly performant.

Suggested change
where = [LITIGATION_ONLY]
params = {}
if courts:
where.append("court_identifier = ANY(:courts)")
params["courts"] = list(courts)
scope = " AND ".join(where)
where = [LITIGATION_ONLY, "created_at >= NOW() - INTERVAL '3 days'"]
params = {}
if courts:
where.append("court_identifier = ANY(:courts)")
params["courts"] = list(courts)
scope = " AND ".join(where)


engine = get_engine(db_url)
try:
with engine.connect() as conn:
# Check if we have cases with null case_type or other expected fields
# suggesting a selector failure (schema drift)
result = conn.execute(
row = conn.execute(
text(
"SELECT count(*) FROM court_cases WHERE case_type IS NULL OR case_type = ''"
)
)
count = result.scalar()
if count > 10: # Threshold for drift detection
f"SELECT count(*) AS total, "
f"count(*) FILTER (WHERE case_type IS NULL OR case_type = '') AS missing "
f"FROM court_cases WHERE {scope}"
),
params,
).one()
total, missing = row.total, row.missing

scope_label = f"courts={list(courts)}" if courts else "all courts"
if total == 0:
print(f"Schema check skipped: no litigation rows in scope ({scope_label})")
return

ratio = missing / total
if ratio > DRIFT_RATIO_THRESHOLD:
print(
f"ERROR: Detected {count} cases with missing case_type. Possible schema drift!"
f"ERROR: {missing}/{total} ({ratio:.1%}) litigation cases missing "
f"case_type ({scope_label}), above {DRIFT_RATIO_THRESHOLD:.0%} "
f"threshold. Possible schema drift!"
)
sys.exit(1)
print("Schema check passed")
print(
f"Schema check passed: {missing}/{total} ({ratio:.2%}) litigation "
f"cases missing case_type ({scope_label})"
)
except Exception as e:
print(f"Schema check failed: {e}")
sys.exit(1)


if __name__ == "__main__":
check()
# Optional positional args scope the check to specific court identifiers
# (e.g. `python scripts/check_schema_drift.py supreme special`).
check(courts=sys.argv[1:] or None)
Loading