From 72bc0462152a5522736331f1edfa7c57ebe0bb05 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Fri, 27 Feb 2026 16:21:49 +0100 Subject: [PATCH] refactor: extract helpers from remaining long functions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extract _start_manual_run and _spawn_background_execution from run_manual route - Extract _upsert_page_publications to deduplicate pagination loop - Extract _classify_attempt from fetch_and_parse_with_retry - Extract _log_alert_threshold_warnings from complete_run_for_user - Functions 3f-3j accepted as kwargs-exception (body ≤50 lines excluding signatures) Co-Authored-By: Claude Opus 4.6 --- app/api/routers/runs.py | 113 +++++++++++++++-------- app/services/ingestion/page_fetch.py | 36 +++++--- app/services/ingestion/pagination.py | 48 +++++++--- app/services/ingestion/run_completion.py | 46 +++++---- 4 files changed, 156 insertions(+), 87 deletions(-) diff --git a/app/api/routers/runs.py b/app/api/routers/runs.py index 687f6ce..28779ef 100644 --- a/app/api/routers/runs.py +++ b/app/api/routers/runs.py @@ -176,6 +176,67 @@ async def cancel_run( ) +async def _start_manual_run( + db_session: AsyncSession, + *, + current_user: User, + ingest_service: ingestion_service.ScholarIngestionService, + idempotency_key: str | None, +) -> tuple[Any, Any, list, dict]: + user_settings = await user_settings_service.get_or_create_settings(db_session, user_id=current_user.id) + run, scholars, start_cstart_map = await ingest_service.initialize_run( + db_session, + user_id=current_user.id, + trigger_type=RunTriggerType.MANUAL, + request_delay_seconds=user_settings.request_delay_seconds, + network_error_retries=settings.ingestion_network_error_retries, + retry_backoff_seconds=settings.ingestion_retry_backoff_seconds, + max_pages_per_scholar=settings.ingestion_max_pages_per_scholar, + page_size=settings.ingestion_page_size, + idempotency_key=idempotency_key, + alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold, + alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold, + alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold, + ) + return user_settings, run, scholars, start_cstart_map + + +def _spawn_background_execution( + ingest_service: ingestion_service.ScholarIngestionService, + *, + run: Any, + current_user: User, + scholars: list, + start_cstart_map: dict, + user_settings: Any, + idempotency_key: str | None, +) -> None: + from app.db.session import get_session_factory + + task = asyncio.create_task( + ingest_service.execute_run( + session_factory=get_session_factory(), + run_id=run.id, + user_id=current_user.id, + scholars=scholars, + start_cstart_map=start_cstart_map, + request_delay_seconds=user_settings.request_delay_seconds, + network_error_retries=settings.ingestion_network_error_retries, + retry_backoff_seconds=settings.ingestion_retry_backoff_seconds, + max_pages_per_scholar=settings.ingestion_max_pages_per_scholar, + page_size=settings.ingestion_page_size, + auto_queue_continuations=settings.ingestion_continuation_queue_enabled, + queue_delay_seconds=settings.ingestion_continuation_base_delay_seconds, + alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold, + alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold, + alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold, + idempotency_key=idempotency_key, + ) + ) + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) + + @router.post( "/manual", response_model=ManualRunEnvelope, @@ -202,52 +263,22 @@ async def run_manual( return reused_payload try: - user_settings = await user_settings_service.get_or_create_settings(db_session, user_id=current_user.id) - - # Initialize run (creates the record and performs safety checks) - run, scholars, start_cstart_map = await ingest_service.initialize_run( + user_settings, run, scholars, start_cstart_map = await _start_manual_run( db_session, - user_id=current_user.id, - trigger_type=RunTriggerType.MANUAL, - request_delay_seconds=user_settings.request_delay_seconds, - network_error_retries=settings.ingestion_network_error_retries, - retry_backoff_seconds=settings.ingestion_retry_backoff_seconds, - max_pages_per_scholar=settings.ingestion_max_pages_per_scholar, - page_size=settings.ingestion_page_size, + current_user=current_user, + ingest_service=ingest_service, idempotency_key=idempotency_key, - alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold, - alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold, - alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold, ) - await db_session.commit() - - # Kick off background execution - from app.db.session import get_session_factory - - task = asyncio.create_task( - ingest_service.execute_run( - session_factory=get_session_factory(), - run_id=run.id, - user_id=current_user.id, - scholars=scholars, - start_cstart_map=start_cstart_map, - request_delay_seconds=user_settings.request_delay_seconds, - network_error_retries=settings.ingestion_network_error_retries, - retry_backoff_seconds=settings.ingestion_retry_backoff_seconds, - max_pages_per_scholar=settings.ingestion_max_pages_per_scholar, - page_size=settings.ingestion_page_size, - auto_queue_continuations=settings.ingestion_continuation_queue_enabled, - queue_delay_seconds=settings.ingestion_continuation_base_delay_seconds, - alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold, - alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold, - alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold, - idempotency_key=idempotency_key, - ) + _spawn_background_execution( + ingest_service, + run=run, + current_user=current_user, + scholars=scholars, + start_cstart_map=start_cstart_map, + user_settings=user_settings, + idempotency_key=idempotency_key, ) - _background_tasks.add(task) - task.add_done_callback(_background_tasks.discard) - return success_payload( request, data={ diff --git a/app/services/ingestion/page_fetch.py b/app/services/ingestion/page_fetch.py index 008cb6b..19aff3f 100644 --- a/app/services/ingestion/page_fetch.py +++ b/app/services/ingestion/page_fetch.py @@ -145,6 +145,24 @@ class PageFetcher: if sleep_seconds > 0: await asyncio.sleep(sleep_seconds) + @staticmethod + def _classify_attempt( + parsed_page: ParsedProfilePage, + *, + network_attempts: int, + rate_limit_attempts: int, + ) -> tuple[int, int, int]: + if parsed_page.state == ParseState.NETWORK_ERROR: + network_attempts += 1 + return network_attempts, rate_limit_attempts, network_attempts + if ( + parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA + and parsed_page.state_reason == "blocked_http_429_rate_limited" + ): + rate_limit_attempts += 1 + return network_attempts, rate_limit_attempts, rate_limit_attempts + return network_attempts, rate_limit_attempts, network_attempts + rate_limit_attempts + 1 + async def fetch_and_parse_with_retry( self, *, @@ -169,19 +187,9 @@ class PageFetcher: page_size=page_size, ) parsed_page = self.parse_page_or_layout_error(fetch_result=fetch_result) - - if parsed_page.state == ParseState.NETWORK_ERROR: - network_attempts += 1 - total_attempts = network_attempts - elif ( - parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA - and parsed_page.state_reason == "blocked_http_429_rate_limited" - ): - rate_limit_attempts += 1 - total_attempts = rate_limit_attempts - else: - total_attempts = network_attempts + rate_limit_attempts + 1 - + network_attempts, rate_limit_attempts, total_attempts = self._classify_attempt( + parsed_page, network_attempts=network_attempts, rate_limit_attempts=rate_limit_attempts + ) attempt_log.append( { "attempt": total_attempts, @@ -192,7 +200,6 @@ class PageFetcher: "fetch_error": fetch_result.error, } ) - if not self._should_retry( parsed_page=parsed_page, network_attempt_count=network_attempts, @@ -201,7 +208,6 @@ class PageFetcher: rate_limit_retries=rate_limit_retries, ): break - await self._sleep_backoff( scholar_id=scholar_id, cstart=cstart, diff --git a/app/services/ingestion/pagination.py b/app/services/ingestion/pagination.py index 9e761b8..0e6b5b0 100644 --- a/app/services/ingestion/pagination.py +++ b/app/services/ingestion/pagination.py @@ -273,6 +273,24 @@ class PaginationEngine: # ── Pagination loop ────────────────────────────────────────────── + @staticmethod + async def _upsert_page_publications( + db_session: Any, + *, + run: CrawlRun, + scholar: ScholarProfile, + publications: list, + seen_canonical: set[str], + state: PagedLoopState, + upsert_publications_fn: Any, + ) -> None: + deduped = _dedupe_publication_candidates(list(publications), seen_canonical=seen_canonical) + if deduped: + discovered_count = await upsert_publications_fn( + db_session, run=run, scholar=scholar, publications=deduped + ) + state.discovered_publication_count += discovered_count + async def _paginate_loop( self, *, @@ -292,14 +310,15 @@ class PaginationEngine: seen_canonical: set[str] = set() if state.parsed_page.publications: - deduped_first = _dedupe_publication_candidates( - list(state.parsed_page.publications), seen_canonical=seen_canonical + await self._upsert_page_publications( + db_session, + run=run, + scholar=scholar, + publications=state.parsed_page.publications, + seen_canonical=seen_canonical, + state=state, + upsert_publications_fn=upsert_publications_fn, ) - if deduped_first: - discovered_count = await upsert_publications_fn( - db_session, run=run, scholar=scholar, publications=deduped_first - ) - state.discovered_publication_count += discovered_count while state.parsed_page.has_show_more_button: await db_session.refresh(run) @@ -337,14 +356,15 @@ class PaginationEngine: ) if next_parsed_page.publications: - deduped_next = _dedupe_publication_candidates( - list(next_parsed_page.publications), seen_canonical=seen_canonical + await self._upsert_page_publications( + db_session, + run=run, + scholar=scholar, + publications=next_parsed_page.publications, + seen_canonical=seen_canonical, + state=state, + upsert_publications_fn=upsert_publications_fn, ) - if deduped_next: - discovered_count = await upsert_publications_fn( - db_session, run=run, scholar=scholar, publications=deduped_next - ) - state.discovered_publication_count += discovered_count if self._handle_page_state_transition(state=state): return diff --git a/app/services/ingestion/run_completion.py b/app/services/ingestion/run_completion.py index b50f770..8f53f6a 100644 --- a/app/services/ingestion/run_completion.py +++ b/app/services/ingestion/run_completion.py @@ -262,25 +262,13 @@ def build_failure_debug_context( return context -def complete_run_for_user( +def _log_alert_threshold_warnings( *, - user_settings: Any, - run: CrawlRun, - scholars: list[ScholarProfile], user_id: int, - progress: RunProgress, - idempotency_key: str | None, - alert_blocked_failure_threshold: int, - alert_network_failure_threshold: int, - alert_retry_scheduled_threshold: int, -) -> tuple[RunFailureSummary, RunAlertSummary]: - failure_summary = summarize_failures(scholar_results=progress.scholar_results) - alert_summary = build_alert_summary( - failure_summary=failure_summary, - alert_blocked_failure_threshold=alert_blocked_failure_threshold, - alert_network_failure_threshold=alert_network_failure_threshold, - alert_retry_scheduled_threshold=alert_retry_scheduled_threshold, - ) + run: CrawlRun, + failure_summary: RunFailureSummary, + alert_summary: RunAlertSummary, +) -> None: if alert_summary.alert_flags["blocked_failure_threshold_exceeded"]: structured_log( logger, @@ -311,6 +299,30 @@ def complete_run_for_user( retries_scheduled_count=failure_summary.retries_scheduled_count, threshold=alert_summary.retry_scheduled_threshold, ) + + +def complete_run_for_user( + *, + user_settings: Any, + run: CrawlRun, + scholars: list[ScholarProfile], + user_id: int, + progress: RunProgress, + idempotency_key: str | None, + alert_blocked_failure_threshold: int, + alert_network_failure_threshold: int, + alert_retry_scheduled_threshold: int, +) -> tuple[RunFailureSummary, RunAlertSummary]: + failure_summary = summarize_failures(scholar_results=progress.scholar_results) + alert_summary = build_alert_summary( + failure_summary=failure_summary, + alert_blocked_failure_threshold=alert_blocked_failure_threshold, + alert_network_failure_threshold=alert_network_failure_threshold, + alert_retry_scheduled_threshold=alert_retry_scheduled_threshold, + ) + _log_alert_threshold_warnings( + user_id=user_id, run=run, failure_summary=failure_summary, alert_summary=alert_summary + ) apply_safety_outcome(user_settings=user_settings, run=run, user_id=user_id, alert_summary=alert_summary) run_status = resolve_run_status( scholar_count=len(scholars),