refactor(scrapers): blend court spiders onto shared bases + fix data-quality bugs - #89
Conversation
…quality bugs Collapse the 8 court spiders (4 listing + 4 enrichment) onto ngm/ngscrape/base_spiders.py (BaseScrapeSpider / BaseCourtCasesSpider / BaseCaseEnrichmentSpider), removing ~1,350 lines of duplication. Per-court HTML parsing stays in the subclasses; the base owns the date-range loop (AD->BS computed once), the per-row guard, bench accumulation + errback, and the single-transaction enrichment save + entity cleaning. Bug fixes folded in: - listing: a single malformed Supreme row no longer aborts the whole day (graceful parties fallback + per-row guard); high/special add a bench errback so one failed bench can't strand a date or leak memory; per-date BS->AD and "today" hoisted out of hot loops. - enrichment: verdict sentinel '**** ** **' no longer stored as '****-**-**'; entity dedup + junk/label/placeholder stripping; supreme caseno parse is '='-safe (parse_qs); high defendant panel parsed independently + address persisted; high label-match fix (trailing-dot / स्थिती spelling) so registration_number/case_status populate; district uses .in_() not LIKE '%dc'; single locked transaction (no double-SELECT / TOCTOU). - court orders: transient FilesPipeline (FileException) downloads are no longer marked permanent (retry counter, escalate after N); the spider no longer green-completes on a swallowed DB error (retry + reconnect, then raise); get_engine uses pool_pre_ping. Schema: add the 4 enrichment columns the code parsed but silently discarded (verdict_type, case_subject, hearing_count, enriched_at). create_all does not ALTER existing tables -- see scripts/migrate_add_enrichment_columns.sql. Tests: first-ever spider/parser tests (none existed) -- tests/unit (small, real Devanagari values) + tests/large (base machinery, enrichment save round-trip, orders failure classification). 147 pass; black + ruff clean. Prepared (not auto-run) SQL: scripts/migrate_add_enrichment_columns.sql and scripts/backfill_data_quality.sql. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
Warning Review limit reached
More reviews will be available in 25 minutes and 10 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (24)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the court-data scrapers by introducing shared base classes (BaseScrapeSpider, BaseCourtCasesSpider, and BaseCaseEnrichmentSpider) to reduce code duplication across district, high, special, and supreme court spiders. It also adds new enrichment columns to the database schema, implements a transient failure retry mechanism in the orders pipeline, and introduces comprehensive unit and integration tests. The code review feedback focuses on preventing potential connection and memory leaks in long-running Scrapy spiders by ensuring that SQLAlchemy sessions are explicitly closed after transactions, during early returns, and when resetting database connections.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| def save_cases(self, data, court_key, date_bs, note=None): | ||
| """Persist cases + hearings for a date and mark the date scraped.""" | ||
| with self.session.begin(): | ||
| for case, hearing in data: | ||
| self.session.merge(case) | ||
| self.session.add(hearing) | ||
| mark_date_scraped(self.session, court_key, date_bs, note) |
There was a problem hiding this comment.
To prevent memory leaks and excessive memory accumulation in long-running Scrapy spiders, it is highly recommended to close the SQLAlchemy session after completing the transaction. Since self.session is reused across multiple dates, calling self.session.close() will clear the session's identity map and release the connection back to the pool, while subsequent operations will transparently open a new transaction/connection when needed.
| def save_cases(self, data, court_key, date_bs, note=None): | |
| """Persist cases + hearings for a date and mark the date scraped.""" | |
| with self.session.begin(): | |
| for case, hearing in data: | |
| self.session.merge(case) | |
| self.session.add(hearing) | |
| mark_date_scraped(self.session, court_key, date_bs, note) | |
| def save_cases(self, data, court_key, date_bs, note=None): | |
| """Persist cases + hearings for a date and mark the date scraped.""" | |
| try: | |
| with self.session.begin(): | |
| for case, hearing in data: | |
| self.session.merge(case) | |
| self.session.add(hearing) | |
| mark_date_scraped(self.session, court_key, date_bs, note) | |
| finally: | |
| self.session.close() |
| def save_enrichment( | ||
| self, | ||
| case_number, | ||
| court_identifier, | ||
| core_fields, | ||
| extra_updates, | ||
| entities, | ||
| ): | ||
| """Apply parsed enrichment in a single locked transaction. | ||
|
|
||
| Returns ``True`` if the row is in the enriched state afterwards (saved now | ||
| or already enriched by a concurrent worker), ``False`` only if the case is | ||
| missing from the DB. | ||
| """ | ||
| now = self._now_ktm() | ||
| with self.session.begin(): | ||
| case = self._get_case(case_number, court_identifier, lock=True) | ||
| if not case: | ||
| self.logger.error(f"Case {case_number} not found for enrichment") | ||
| return False | ||
|
|
||
| if case.status == "enriched": | ||
| self.logger.info(f"Case {case_number} already enriched, skipping") | ||
| return True | ||
|
|
||
| for key, value in core_fields.items(): | ||
| setattr(case, key, value) | ||
|
|
||
| if case.extra_data is None: | ||
| case.extra_data = {} | ||
| case.extra_data.update(extra_updates) | ||
| flag_modified(case, "extra_data") | ||
|
|
||
| case.status = "enriched" | ||
| case.enriched_at = now | ||
| case.updated_at = now | ||
|
|
||
| self._replace_entities(case_number, court_identifier, entities, now) | ||
| return True | ||
|
|
There was a problem hiding this comment.
Similar to save_cases, save_enrichment is called repeatedly across thousands of cases. To avoid memory leaks from the accumulation of CourtCase and CaseEntity objects in the SQLAlchemy session's identity map, ensure the session is closed at the end of the transaction.
def save_enrichment(
self,
case_number,
court_identifier,
core_fields,
extra_updates,
entities,
):
"""Apply parsed enrichment in a single locked transaction.
Returns ``True`` if the row is in the enriched state afterwards (saved now
or already enriched by a concurrent worker), ``False`` only if the case is
missing from the DB.
"""
now = self._now_ktm()
try:
with self.session.begin():
case = self._get_case(case_number, court_identifier, lock=True)
if not case:
self.logger.error(f"Case {case_number} not found for enrichment")
return False
if case.status == "enriched":
self.logger.info(f"Case {case_number} already enriched, skipping")
return True
for key, value in core_fields.items():
setattr(case, key, value)
if case.extra_data is None:
case.extra_data = {}
case.extra_data.update(extra_updates)
flag_modified(case, "extra_data")
case.status = "enriched"
case.enriched_at = now
case.updated_at = now
self._replace_entities(case_number, court_identifier, entities, now)
return True
finally:
self.session.close()| def _mark_transient(self, spider, case_number, court_identifier, error): | ||
| """Transient failure (download/S3/timeout) — NON-terminal. | ||
|
|
||
| Does not set orders_failed, so the selection query still re-picks the | ||
| case next run. Tracks a retry counter and only escalates to a permanent | ||
| failure after MAX_TRANSIENT_RETRIES so a genuinely-dead URL eventually | ||
| stops being re-queued. | ||
| """ | ||
| try: | ||
| with self.session.begin(): | ||
| case = ( | ||
| self.session.query(CourtCase) | ||
| .filter_by( | ||
| case_number=case_number, court_identifier=court_identifier | ||
| ) | ||
| .first() | ||
| ) | ||
| if not case: | ||
| spider.logger.error( | ||
| f"[{case_number}] Not in DB. Cannot mark transient." | ||
| ) | ||
| return | ||
| if case.extra_data is None: | ||
| case.extra_data = {} | ||
|
|
||
| retries = int(case.extra_data.get("orders_transient_retries", 0)) + 1 | ||
| case.extra_data["orders_transient_retries"] = retries | ||
| case.extra_data["orders_transient_error"] = error | ||
| case.extra_data["orders_transient_at"] = self._now_iso() | ||
| # A transient path must never leave a stale permanent flag. | ||
| case.extra_data.pop("orders_failed", None) | ||
| case.extra_data.pop("orders_error", None) | ||
| case.extra_data.pop("orders_failed_at", None) | ||
|
|
||
| if retries >= self.MAX_TRANSIENT_RETRIES: | ||
| case.extra_data["orders_failed"] = True | ||
| case.extra_data["orders_error"] = ( | ||
| f"transient_exhausted after {retries} retries: {error}" | ||
| ) | ||
| case.extra_data["orders_failed_at"] = self._now_iso() | ||
| spider.logger.error( | ||
| f"[{case_number}] Transient retries exhausted ({retries}) " | ||
| "— marking permanent." | ||
| ) | ||
| else: | ||
| spider.logger.warning( | ||
| f"[{case_number}] Transient download failure " | ||
| f"(retry {retries}/{self.MAX_TRANSIENT_RETRIES}) — " | ||
| "will retry next run." | ||
| ) | ||
| flag_modified(case, "extra_data") | ||
| except Exception: | ||
| spider.logger.exception(f"[{case_number}] Error marking transient") | ||
| raise |
There was a problem hiding this comment.
To prevent connection and memory leaks in the SupremeCourtOrdersPipeline when handling transient errors, explicitly close the SQLAlchemy session in a finally block after the transaction completes.
def _mark_transient(self, spider, case_number, court_identifier, error):
"""Transient failure (download/S3/timeout) — NON-terminal.
Does not set orders_failed, so the selection query still re-picks the
case next run. Tracks a retry counter and only escalates to a permanent
failure after MAX_TRANSIENT_RETRIES so a genuinely-dead URL eventually
stops being re-queued.
"""
try:
with self.session.begin():
case = (
self.session.query(CourtCase)
.filter_by(
case_number=case_number, court_identifier=court_identifier
)
.first()
)
if not case:
spider.logger.error(
f"[{case_number}] Not in DB. Cannot mark transient."
)
return
if case.extra_data is None:
case.extra_data = {}
retries = int(case.extra_data.get("orders_transient_retries", 0)) + 1
case.extra_data["orders_transient_retries"] = retries
case.extra_data["orders_transient_error"] = error
case.extra_data["orders_transient_at"] = self._now_iso()
# A transient path must never leave a stale permanent flag.
case.extra_data.pop("orders_failed", None)
case.extra_data.pop("orders_error", None)
case.extra_data.pop("orders_failed_at", None)
if retries >= self.MAX_TRANSIENT_RETRIES:
case.extra_data["orders_failed"] = True
case.extra_data["orders_error"] = (
f"transient_exhausted after {retries} retries: {error}"
)
case.extra_data["orders_failed_at"] = self._now_iso()
spider.logger.error(
f"[{case_number}] Transient retries exhausted ({retries}) "
"— marking permanent."
)
else:
spider.logger.warning(
f"[{case_number}] Transient download failure "
f"(retry {retries}/{self.MAX_TRANSIENT_RETRIES}) — "
"will retry next run."
)
flag_modified(case, "extra_data")
except Exception:
spider.logger.exception(f"[{case_number}] Error marking transient")
raise
finally:
self.session.close()| if self.backfill_case_type: | ||
| # If a parallel worker already enriched this row, only backfill the | ||
| # missing case_type (don't rebuild entities/hearings). | ||
| with self.session.begin(): | ||
| case = self._get_case(case_number, code_name, lock=True) | ||
| if case and case.status == "enriched": | ||
| if case.case_type: | ||
| return | ||
| case_type = enrichment_data.get("case_type") | ||
| if case_type: | ||
| case.case_type = case_type[:200] | ||
| case.updated_at = self._now_ktm() | ||
| self.logger.info( | ||
| f"Backfilled case_type for {case_number} ({code_name})" | ||
| ) | ||
| return |
There was a problem hiding this comment.
When returning early during the backfill_case_type check, the SQLAlchemy session is left open with the loaded CourtCase object in its identity map. Wrap this block in a try...finally to ensure the session is closed and resources are freed, even on early returns.
| if self.backfill_case_type: | |
| # If a parallel worker already enriched this row, only backfill the | |
| # missing case_type (don't rebuild entities/hearings). | |
| with self.session.begin(): | |
| case = self._get_case(case_number, code_name, lock=True) | |
| if case and case.status == "enriched": | |
| if case.case_type: | |
| return | |
| case_type = enrichment_data.get("case_type") | |
| if case_type: | |
| case.case_type = case_type[:200] | |
| case.updated_at = self._now_ktm() | |
| self.logger.info( | |
| f"Backfilled case_type for {case_number} ({code_name})" | |
| ) | |
| return | |
| if self.backfill_case_type: | |
| # If a parallel worker already enriched this row, only backfill the | |
| # missing case_type (don't rebuild entities/hearings). | |
| try: | |
| with self.session.begin(): | |
| case = self._get_case(case_number, code_name, lock=True) | |
| if case and case.status == "enriched": | |
| if case.case_type: | |
| return | |
| case_type = enrichment_data.get("case_type") | |
| if case_type: | |
| case.case_type = case_type[:200] | |
| case.updated_at = self._now_ktm() | |
| self.logger.info( | |
| f"Backfilled case_type for {case_number} ({code_name})" | |
| ) | |
| return | |
| finally: | |
| self.session.close() |
| try: | ||
| self.engine.dispose() | ||
| self.session = get_session(self.engine) | ||
| except Exception: | ||
| self.logger.exception("Failed to reset DB session before retry") |
There was a problem hiding this comment.
When resetting the database connection during a retry attempt, the old self.session is abandoned without being closed. This can leak database connections or leave them in an active state on the server. Explicitly close the old session before disposing of the engine and creating a new session.
| try: | |
| self.engine.dispose() | |
| self.session = get_session(self.engine) | |
| except Exception: | |
| self.logger.exception("Failed to reset DB session before retry") | |
| try: | |
| if hasattr(self, "session") and self.session: | |
| self.session.close() | |
| self.engine.dispose() | |
| self.session = get_session(self.engine) | |
| except Exception: | |
| self.logger.exception("Failed to reset DB session before retry") |
… on missing detail link Addresses the PR review (Gemini + self-review): - base save_cases / save_enrichment, pipeline _mark_transient, district backfill block, and the orders retry now close the SQLAlchemy session in a finally so the identity map / connections don't accumulate over a long crawl. - correct the BaseCaseEnrichmentSpider docstring: re-enrichment fully replaces parties and does NOT preserve nes_id (the preservation was intentionally dropped — the docstring claimed otherwise). - supreme enrichment: a missing detail-link / caseno no longer permanently marks the case failed (restores the original "leave pending, retry" behavior; the mark_failed was an unintended behavior change). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Addressed review + ran a prod-safe dry runReview comments — fixed in
Dry run — local SQLite only, zero prod writes (hard-gated to refuse anything but
No exceptions from spider code in either run, and the new session-close behavior was exercised under thousands of real saves without issue. |
1 participant
Summary
Blends the 8 court spiders (4
*_court_caseslisting + 4*_case_enrichment) onto a shared base (ngm/ngscrape/base_spiders.py), removing ~1,350 lines of duplication (+1042/−2392 across the 8 spiders + models), while folding in correctness + data-quality fixes surfaced by a full code review and a live-DB audit. Adds the first-ever spider test suite.Per-court HTML parsing stays in the subclasses; the base owns the shared skeleton: the date-range loop (AD→BS computed once — a big perf win for district's 77 courts), the per-row guard, bench accumulation + errback, and the single-transaction enrichment save + entity cleaning.
Correctness / data-quality fixes
Listing
raises and discards the entire day (graceful fallback + per-row guard).errback, so one failed bench can't strand the date forever or leak accumulated rows in memory.Enrichment
**** ** **is no longer stored as the fake date****-**-**(this had polluted ~140k district rows — see backfill).CaseEntityrebuild now dedupes and strips header-label / placeholder / whitespace-padded names.casenoparse is=-safe (parse_qs)..and use theस्थितीspelling) soregistration_number/case_statuspopulate again..in_(codes)instead of a non-sargableLIKE '%dc'.Court orders
FilesPipeline(FileException) download failures are no longer persisted as permanentorders_failed(retry counter, escalate only after N) — this stranded 13 decided cases that likely do have downloadable orders._get_cases_to_scrape()retries with a fresh connection and raises loudly if it can't recover (the silentexcept: returnis why the CronJob reportedTotal: 0for ~40 days).get_engine()now usespool_pre_ping=True.Schema
Adds the 4 columns the enrichment code parses but which never existed, so the values were silently discarded:
verdict_type,case_subject,hearing_count,enriched_at.create_alldoes not ALTER existing tables — the prod migration isscripts/migrate_add_enrichment_columns.sqland must run before this image deploys.Tests
No spider/parser tests existed before. Adds
tests/unit/(small, real Devanagari examples) +tests/large/(base machinery, enrichment save round-trip, orders failure classification) on an in-memory SQLite harness (tests/conftest.py, JSONB→JSON). 147 pass;scripts/format.sh --check(black + ruff) clean.Follow-ups (not in this PR)
scripts/migrate_add_enrichment_columns.sql(columns + indexes) onngm_v1before deploy.scripts/backfill_data_quality.sqlafter review (verdict sentinel → NULL; HC fields recovered fromextra_data, no re-scrape; supreme judge_names; release the 13 transient orders).DATABASE_URLand restart thengm-supreme-court-ordersCronJob.🤖 Generated with Claude Code