diff --git a/app/api/routers/admin.py b/app/api/routers/admin.py index 9844c3e..b1f99b8 100644 --- a/app/api/routers/admin.py +++ b/app/api/routers/admin.py @@ -20,6 +20,7 @@ from app.auth.deps import get_auth_service from app.auth.service import AuthService from app.db.models import User from app.db.session import get_db_session +from app.logging_utils import structured_log from app.services.domains.users import application as user_service logger = logging.getLogger(__name__) @@ -48,14 +49,7 @@ async def list_users( admin_user: User = Depends(get_api_admin_user), ): users = await user_service.list_users(db_session) - logger.info( - "api.admin.users_listed", - extra={ - "event": "api.admin.users_listed", - "admin_user_id": int(admin_user.id), - "user_count": len(users), - }, - ) + structured_log(logger, "info", "api.admin.users_listed", admin_user_id=int(admin_user.id), user_count=len(users)) return success_payload( request, data={ @@ -92,15 +86,7 @@ async def create_user( message=str(exc), ) from exc - logger.info( - "api.admin.user_created", - extra={ - "event": "api.admin.user_created", - "admin_user_id": int(admin_user.id), - "target_user_id": int(created_user.id), - "target_is_admin": bool(created_user.is_admin), - }, - ) + structured_log(logger, "info", "api.admin.user_created", admin_user_id=int(admin_user.id), target_user_id=int(created_user.id), target_is_admin=bool(created_user.is_admin)) return success_payload( request, data=_serialize_user(created_user), @@ -140,15 +126,7 @@ async def set_user_active( user=target_user, is_active=bool(payload.is_active), ) - logger.info( - "api.admin.user_active_updated", - extra={ - "event": "api.admin.user_active_updated", - "admin_user_id": int(admin_user.id), - "target_user_id": int(updated_user.id), - "is_active": bool(updated_user.is_active), - }, - ) + structured_log(logger, "info", "api.admin.user_active_updated", admin_user_id=int(admin_user.id), target_user_id=int(updated_user.id), is_active=bool(updated_user.is_active)) return success_payload( request, data=_serialize_user(updated_user), @@ -188,14 +166,7 @@ async def reset_user_password( user=target_user, password_hash=auth_service.hash_password(validated_password), ) - logger.info( - "api.admin.user_password_reset", - extra={ - "event": "api.admin.user_password_reset", - "admin_user_id": int(admin_user.id), - "target_user_id": int(target_user.id), - }, - ) + structured_log(logger, "info", "api.admin.user_password_reset", admin_user_id=int(admin_user.id), target_user_id=int(target_user.id)) return success_payload( request, data={"message": f"Password reset: {target_user.email}"}, diff --git a/app/api/routers/admin_dbops.py b/app/api/routers/admin_dbops.py index be23092..8230ab9 100644 --- a/app/api/routers/admin_dbops.py +++ b/app/api/routers/admin_dbops.py @@ -21,6 +21,7 @@ from app.api.schemas import ( ) from app.db.models import DataRepairJob, User from app.db.session import get_db_session +from app.logging_utils import structured_log from app.services.domains.dbops import ( collect_integrity_report, list_repair_jobs, @@ -135,16 +136,7 @@ async def get_integrity_report( admin_user: User = Depends(get_api_admin_user), ): report = await collect_integrity_report(db_session) - logger.info( - "api.admin.db.integrity_checked", - extra={ - "event": "api.admin.db.integrity_checked", - "admin_user_id": int(admin_user.id), - "status": report.get("status"), - "failure_count": len(report.get("failures", [])), - "warning_count": len(report.get("warnings", [])), - }, - ) + structured_log(logger, "info", "api.admin.db.integrity_checked", admin_user_id=int(admin_user.id), status=report.get("status"), failure_count=len(report.get("failures", [])), warning_count=len(report.get("warnings", []))) return success_payload(request, data=report) @@ -159,15 +151,7 @@ async def get_repair_jobs( admin_user: User = Depends(get_api_admin_user), ): jobs = await list_repair_jobs(db_session, limit=limit) - logger.info( - "api.admin.db.repair_jobs_listed", - extra={ - "event": "api.admin.db.repair_jobs_listed", - "admin_user_id": int(admin_user.id), - "limit": int(limit), - "job_count": len(jobs), - }, - ) + structured_log(logger, "info", "api.admin.db.repair_jobs_listed", admin_user_id=int(admin_user.id), limit=int(limit), job_count=len(jobs)) return success_payload( request, data={"jobs": [_serialize_repair_job(job) for job in jobs]}, @@ -201,18 +185,15 @@ async def get_pdf_queue( offset=resolved_offset, status=normalized_status, ) - logger.info( - "api.admin.db.pdf_queue_listed", - extra={ - "event": "api.admin.db.pdf_queue_listed", - "admin_user_id": int(admin_user.id), - "page": int(resolved_page), - "page_size": int(resolved_limit), - "offset": int(resolved_offset), - "status": normalized_status, - "item_count": len(queue_page.items), - "total_count": int(queue_page.total_count), - }, + structured_log( + logger, "info", "api.admin.db.pdf_queue_listed", + admin_user_id=int(admin_user.id), + page=int(resolved_page), + page_size=int(resolved_limit), + offset=int(resolved_offset), + status=normalized_status, + item_count=len(queue_page.items), + total_count=int(queue_page.total_count), ) return success_payload( request, @@ -251,15 +232,7 @@ async def requeue_pdf_lookup( message="Publication not found.", ) status, message = _requeue_response_state(queued=result.queued) - logger.info( - "api.admin.db.pdf_queue_requeue_requested", - extra={ - "event": "api.admin.db.pdf_queue_requeue_requested", - "admin_user_id": int(admin_user.id), - "publication_id": int(publication_id), - "queued": bool(result.queued), - }, - ) + structured_log(logger, "info", "api.admin.db.pdf_queue_requeue_requested", admin_user_id=int(admin_user.id), publication_id=int(publication_id), queued=bool(result.queued)) return success_payload( request, data={ @@ -287,16 +260,7 @@ async def requeue_all_missing_pdfs( request_email=admin_user.email, limit=limit, ) - logger.info( - "api.admin.db.pdf_queue_requeue_all", - extra={ - "event": "api.admin.db.pdf_queue_requeue_all", - "admin_user_id": int(admin_user.id), - "limit": int(limit), - "requested_count": int(result.requested_count), - "queued_count": int(result.queued_count), - }, - ) + structured_log(logger, "info", "api.admin.db.pdf_queue_requeue_all", admin_user_id=int(admin_user.id), limit=int(limit), requested_count=int(result.requested_count), queued_count=int(result.queued_count)) return success_payload( request, data={ @@ -336,18 +300,15 @@ async def trigger_publication_link_repair( code="invalid_repair_scope", message=str(exc), ) from exc - logger.info( - "api.admin.db.publication_link_repair_triggered", - extra={ - "event": "api.admin.db.publication_link_repair_triggered", - "admin_user_id": int(admin_user.id), - "scope_mode": payload.scope_mode, - "target_user_id": int(payload.user_id) if payload.user_id is not None else None, - "dry_run": bool(payload.dry_run), - "gc_orphan_publications": bool(payload.gc_orphan_publications), - "job_id": int(result["job_id"]), - "status": result["status"], - }, + structured_log( + logger, "info", "api.admin.db.publication_link_repair_triggered", + admin_user_id=int(admin_user.id), + scope_mode=payload.scope_mode, + target_user_id=int(payload.user_id) if payload.user_id is not None else None, + dry_run=bool(payload.dry_run), + gc_orphan_publications=bool(payload.gc_orphan_publications), + job_id=int(result["job_id"]), + status=result["status"], ) return success_payload(request, data=result) @@ -379,17 +340,7 @@ async def trigger_publication_near_duplicate_repair( code="invalid_near_duplicate_repair_request", message=str(exc), ) from exc - logger.info( - "api.admin.db.publication_near_duplicate_repair_triggered", - extra={ - "event": "api.admin.db.publication_near_duplicate_repair_triggered", - "admin_user_id": int(admin_user.id), - "dry_run": bool(payload.dry_run), - "selected_cluster_count": len(payload.selected_cluster_keys), - "job_id": int(result["job_id"]), - "status": result["status"], - }, - ) + structured_log(logger, "info", "api.admin.db.publication_near_duplicate_repair_triggered", admin_user_id=int(admin_user.id), dry_run=bool(payload.dry_run), selected_cluster_count=len(payload.selected_cluster_keys), job_id=int(result["job_id"]), status=result["status"]) return success_payload(request, data=result) @@ -435,15 +386,7 @@ async def drop_all_publications( ) await db_session.commit() - logger.warning( - "api.admin.db.all_publications_dropped", - extra={ - "event": "api.admin.db.all_publications_dropped", - "admin_user_id": int(admin_user.id), - "admin_email": admin_user.email, - "deleted_count": int(total_publications), - }, - ) + structured_log(logger, "warning", "api.admin.db.all_publications_dropped", admin_user_id=int(admin_user.id), admin_email=admin_user.email, deleted_count=int(total_publications)) return success_payload( request, data={ diff --git a/app/api/routers/auth.py b/app/api/routers/auth.py index 8f8da17..3faffa6 100644 --- a/app/api/routers/auth.py +++ b/app/api/routers/auth.py @@ -17,6 +17,7 @@ from app.api.schemas import ( MessageEnvelope, ) from app.auth.deps import get_auth_service, get_login_rate_limiter +from app.logging_utils import structured_log from app.auth.rate_limit import SlidingWindowRateLimiter from app.auth import runtime as auth_runtime from app.auth.service import AuthService @@ -36,14 +37,7 @@ def _login_limiter_key_and_email(request: Request, payload: LoginRequest) -> tup def _raise_rate_limited(normalized_email: str, retry_after_seconds: int) -> None: - logger.warning( - "api.auth.login_rate_limited", - extra={ - "event": "api.auth.login_rate_limited", - "email": normalized_email, - "retry_after_seconds": retry_after_seconds, - }, - ) + structured_log(logger, "warning", "api.auth.login_rate_limited", email=normalized_email, retry_after_seconds=retry_after_seconds) raise ApiException( status_code=429, code="rate_limited", @@ -62,10 +56,7 @@ def _serialize_user_payload(user: User) -> dict[str, object]: def _raise_invalid_credentials(*, normalized_email: str) -> None: - logger.info( - "api.auth.login_failed", - extra={"event": "api.auth.login_failed", "email": normalized_email}, - ) + structured_log(logger, "info", "api.auth.login_failed", email=normalized_email) raise ApiException( status_code=401, code="invalid_credentials", @@ -105,14 +96,7 @@ async def login( email=user.email, is_admin=user.is_admin, ) - logger.info( - "api.auth.login_succeeded", - extra={ - "event": "api.auth.login_succeeded", - "user_id": user.id, - "is_admin": user.is_admin, - }, - ) + structured_log(logger, "info", "api.auth.login_succeeded", user_id=user.id, is_admin=user.is_admin) return success_payload( request, data={ @@ -199,13 +183,7 @@ async def change_password( user=current_user, password_hash=auth_service.hash_password(validated_password), ) - logger.info( - "api.auth.password_changed", - extra={ - "event": "api.auth.password_changed", - "user_id": int(current_user.id), - }, - ) + structured_log(logger, "info", "api.auth.password_changed", user_id=int(current_user.id)) return success_payload( request, data={"message": "Password updated successfully."}, @@ -222,13 +200,7 @@ async def logout( ): current_user = await auth_runtime.get_authenticated_user(request, db_session) auth_runtime.invalidate_session(request) - logger.info( - "api.auth.logout", - extra={ - "event": "api.auth.logout", - "user_id": int(current_user.id) if current_user is not None else None, - }, - ) + structured_log(logger, "info", "api.auth.logout", user_id=int(current_user.id) if current_user is not None else None) return success_payload( request, data={ diff --git a/app/api/routers/publications.py b/app/api/routers/publications.py index 7de2f12..7048d6b 100644 --- a/app/api/routers/publications.py +++ b/app/api/routers/publications.py @@ -22,6 +22,7 @@ from app.api.schemas import ( ) from app.db.models import User from app.db.session import get_db_session +from app.logging_utils import structured_log from app.services.domains.publication_identifiers import application as identifier_service from app.services.domains.publications import application as publication_service from app.services.domains.scholars import application as scholar_service @@ -340,30 +341,6 @@ async def _favorite_publication_state( return identifiers[0] if identifiers else current -def _log_retry_pdf_result( - *, - current_user: User, - scholar_profile_id: int, - publication_id: int, - queued: bool, - resolved_pdf: bool, - pdf_status: str, - doi: str | None, -) -> None: - logger.info( - "api.publications.retry_pdf", - extra={ - "event": "api.publications.retry_pdf", - "user_id": current_user.id, - "scholar_profile_id": scholar_profile_id, - "publication_id": publication_id, - "queued": queued, - "resolved_pdf": resolved_pdf, - "pdf_status": pdf_status, - }, - ) - - @router.get( "", response_model=PublicationsListEnvelope, @@ -443,14 +420,7 @@ async def mark_all_publications_read( db_session, user_id=current_user.id, ) - logger.info( - "api.publications.mark_all_read", - extra={ - "event": "api.publications.mark_all_read", - "user_id": current_user.id, - "updated_count": updated_count, - }, - ) + structured_log(logger, "info", "api.publications.mark_all_read", user_id=current_user.id, updated_count=updated_count) return success_payload( request, data={ @@ -481,15 +451,7 @@ async def mark_selected_publications_read( user_id=current_user.id, selections=selection_pairs, ) - logger.info( - "api.publications.mark_selected_read", - extra={ - "event": "api.publications.mark_selected_read", - "user_id": current_user.id, - "requested_count": len(selection_pairs), - "updated_count": updated_count, - }, - ) + structured_log(logger, "info", "api.publications.mark_selected_read", user_id=current_user.id, requested_count=len(selection_pairs), updated_count=updated_count) return success_payload( request, data={ @@ -523,14 +485,14 @@ async def retry_publication_pdf( resolved_pdf=resolved_pdf, pdf_status=current.pdf_status, ) - _log_retry_pdf_result( - current_user=current_user, + structured_log( + logger, "info", "api.publications.retry_pdf", + user_id=current_user.id, scholar_profile_id=payload.scholar_profile_id, publication_id=publication_id, queued=queued, resolved_pdf=resolved_pdf, pdf_status=current.pdf_status, - doi=None, ) return success_payload( request, @@ -561,16 +523,7 @@ async def toggle_publication_favorite( publication_id=publication_id, is_favorite=payload.is_favorite, ) - logger.info( - "api.publications.favorite", - extra={ - "event": "api.publications.favorite", - "user_id": current_user.id, - "scholar_profile_id": payload.scholar_profile_id, - "publication_id": publication_id, - "is_favorite": bool(payload.is_favorite), - }, - ) + structured_log(logger, "info", "api.publications.favorite", user_id=current_user.id, scholar_profile_id=payload.scholar_profile_id, publication_id=publication_id, is_favorite=bool(payload.is_favorite)) return success_payload( request, data={ diff --git a/app/api/routers/runs.py b/app/api/routers/runs.py index b03a45d..f6e6644 100644 --- a/app/api/routers/runs.py +++ b/app/api/routers/runs.py @@ -22,6 +22,7 @@ from app.api.schemas import ( ) from app.db.models import RunStatus, RunTriggerType, User from app.db.session import get_db_session +from app.logging_utils import structured_log from app.services.domains.ingestion import application as ingestion_service from app.services.domains.ingestion import safety as run_safety_service from app.services.domains.runs import application as run_service @@ -287,31 +288,21 @@ async def _load_safety_state( if run_safety_service.clear_expired_cooldown(user_settings): await db_session.commit() await db_session.refresh(user_settings) - logger.info( - "api.runs.safety_cooldown_cleared", - extra={ - "event": "api.runs.safety_cooldown_cleared", - "user_id": user_id, - "reason": previous_safety_state.get("cooldown_reason"), - "cooldown_until": previous_safety_state.get("cooldown_until"), - "metric_name": "api_runs_safety_cooldown_cleared_total", - "metric_value": 1, - }, - ) + structured_log( + logger, "info", "api.runs.safety_cooldown_cleared", + user_id=user_id, + reason=previous_safety_state.get("cooldown_reason"), + cooldown_until=previous_safety_state.get("cooldown_until"), + ) return run_safety_service.get_safety_state_payload(user_settings) def _raise_manual_runs_disabled(*, user_id: int, safety_state: dict[str, Any]) -> None: - logger.warning( - "api.runs.manual_blocked_policy", - extra={ - "event": "api.runs.manual_blocked_policy", - "user_id": user_id, - "policy": {"manual_run_allowed": False}, - "safety_state": safety_state, - "metric_name": "api_runs_manual_blocked_policy_total", - "metric_value": 1, - }, + structured_log( + logger, "warning", "api.runs.manual_blocked_policy", + user_id=user_id, + policy={"manual_run_allowed": False}, + safety_state=safety_state, ) raise ApiException( status_code=403, @@ -399,7 +390,7 @@ async def _recover_integrity_error( if idempotency_key is None: logger.exception( "api.runs.manual_integrity_error", - extra={"event": "api.runs.manual_integrity_error", "user_id": user_id}, + extra={"user_id": user_id}, ) raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc existing_run = await run_service.get_manual_run_by_idempotency_key( @@ -410,7 +401,7 @@ async def _recover_integrity_error( if existing_run is None: logger.exception( "api.runs.manual_integrity_error", - extra={"event": "api.runs.manual_integrity_error", "user_id": user_id}, + extra={"user_id": user_id}, ) raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc if existing_run.status in (RunStatus.RUNNING, RunStatus.RESOLVING): @@ -471,17 +462,12 @@ async def _execute_manual_run( def _raise_manual_blocked_safety(*, exc, user_id: int) -> None: - logger.info( - "api.runs.manual_blocked_safety", - extra={ - "event": "api.runs.manual_blocked_safety", - "user_id": user_id, - "reason": exc.safety_state.get("cooldown_reason"), - "cooldown_until": exc.safety_state.get("cooldown_until"), - "cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"), - "metric_name": "api_runs_manual_blocked_safety_total", - "metric_value": 1, - }, + structured_log( + logger, "info", "api.runs.manual_blocked_safety", + user_id=user_id, + reason=exc.safety_state.get("cooldown_reason"), + cooldown_until=exc.safety_state.get("cooldown_until"), + cooldown_remaining_seconds=exc.safety_state.get("cooldown_remaining_seconds"), ) raise ApiException( status_code=429, @@ -494,7 +480,7 @@ def _raise_manual_blocked_safety(*, exc, user_id: int) -> None: def _raise_manual_failed(*, exc: Exception, user_id: int) -> None: logger.exception( "api.runs.manual_failed", - extra={"event": "api.runs.manual_failed", "user_id": user_id}, + extra={"user_id": user_id}, ) raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from exc diff --git a/app/api/routers/scholars.py b/app/api/routers/scholars.py index 2dc0e2f..f0784fe 100644 --- a/app/api/routers/scholars.py +++ b/app/api/routers/scholars.py @@ -23,6 +23,7 @@ from app.api.schemas import ( ) from app.db.models import User from app.db.session import get_db_session +from app.logging_utils import structured_log from app.services.domains.ingestion import queue as ingestion_queue_service from app.services.domains.portability import application as import_export_service from app.services.domains.scholar import rate_limit as scholar_rate_limit @@ -80,24 +81,18 @@ async def _enqueue_initial_scrape_job_for_scholar( await db_session.commit() except Exception: await db_session.rollback() - logger.warning( - "api.scholars.initial_scrape_enqueue_failed", - extra={ - "event": "api.scholars.initial_scrape_enqueue_failed", - "user_id": user_id, - "scholar_profile_id": profile.id, - }, + structured_log( + logger, "warning", "api.scholars.initial_scrape_enqueue_failed", + user_id=user_id, + scholar_profile_id=profile.id, ) return False - logger.info( - "api.scholars.initial_scrape_enqueued", - extra={ - "event": "api.scholars.initial_scrape_enqueued", - "user_id": user_id, - "scholar_profile_id": profile.id, - "reason": INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON, - }, + structured_log( + logger, "info", "api.scholars.initial_scrape_enqueued", + user_id=user_id, + scholar_profile_id=profile.id, + reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON, ) return True @@ -143,15 +138,12 @@ async def _hydrate_scholar_metadata_if_needed( should_skip, remaining_seconds = _is_create_hydration_rate_limited() if should_skip: - logger.info( - "api.scholars.create_metadata_hydration_skipped", - extra={ - "event": "api.scholars.create_metadata_hydration_skipped", - "reason": "scholar_request_throttle_active", - "user_id": user_id, - "scholar_profile_id": profile.id, - "retry_after_seconds": round(remaining_seconds, 3), - }, + structured_log( + logger, "info", "api.scholars.create_metadata_hydration_skipped", + reason="scholar_request_throttle_active", + user_id=user_id, + scholar_profile_id=profile.id, + retry_after_seconds=round(remaining_seconds, 3), ) return profile @@ -165,23 +157,17 @@ async def _hydrate_scholar_metadata_if_needed( timeout=CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS, ) except TimeoutError: - logger.info( - "api.scholars.create_metadata_hydration_skipped", - extra={ - "event": "api.scholars.create_metadata_hydration_skipped", - "reason": "create_timeout", - "user_id": user_id, - "scholar_profile_id": profile.id, - }, + structured_log( + logger, "info", "api.scholars.create_metadata_hydration_skipped", + reason="create_timeout", + user_id=user_id, + scholar_profile_id=profile.id, ) except Exception: - logger.warning( - "api.scholars.create_metadata_hydration_failed", - extra={ - "event": "api.scholars.create_metadata_hydration_failed", - "user_id": user_id, - "scholar_profile_id": profile.id, - }, + structured_log( + logger, "warning", "api.scholars.create_metadata_hydration_failed", + user_id=user_id, + scholar_profile_id=profile.id, ) return profile @@ -330,14 +316,7 @@ async def import_scholars_and_publications( code="invalid_import_payload", message=str(exc), ) from exc - logger.info( - "api.scholars.imported", - extra={ - "event": "api.scholars.imported", - "user_id": current_user.id, - **result, - }, - ) + structured_log(logger, "info", "api.scholars.imported", user_id=current_user.id, **result) return success_payload(request, data=result) @@ -367,14 +346,7 @@ async def create_scholar( code="invalid_scholar", message=str(exc), ) from exc - logger.info( - "api.scholars.created", - extra={ - "event": "api.scholars.created", - "user_id": current_user.id, - "scholar_profile_id": created.id, - }, - ) + structured_log(logger, "info", "api.scholars.created", user_id=current_user.id, scholar_profile_id=created.id) did_queue_initial_scrape = await _enqueue_initial_scrape_job_for_scholar( db_session, profile=created, @@ -421,16 +393,7 @@ async def search_scholars( message=str(exc), ) from exc - logger.info( - "api.scholars.search.completed", - extra={ - "event": "api.scholars.search.completed", - "user_id": current_user.id, - "query": query.strip(), - "candidate_count": len(parsed.candidates), - "state": parsed.state.value, - }, - ) + structured_log(logger, "info", "api.scholars.search.completed", user_id=current_user.id, query=query.strip(), candidate_count=len(parsed.candidates), state=parsed.state.value) return success_payload( request, data=_search_response_data(query, parsed), @@ -459,15 +422,7 @@ async def toggle_scholar( message="Scholar not found.", ) updated = await scholar_service.toggle_scholar_enabled(db_session, profile=profile) - logger.info( - "api.scholars.toggled", - extra={ - "event": "api.scholars.toggled", - "user_id": current_user.id, - "scholar_profile_id": updated.id, - "is_enabled": updated.is_enabled, - }, - ) + structured_log(logger, "info", "api.scholars.toggled", user_id=current_user.id, scholar_profile_id=updated.id, is_enabled=updated.is_enabled) return success_payload( request, data=_serialize_scholar(updated), @@ -500,14 +455,7 @@ async def delete_scholar( profile=profile, upload_dir=settings.scholar_image_upload_dir, ) - logger.info( - "api.scholars.deleted", - extra={ - "event": "api.scholars.deleted", - "user_id": current_user.id, - "scholar_profile_id": scholar_profile_id, - }, - ) + structured_log(logger, "info", "api.scholars.deleted", user_id=current_user.id, scholar_profile_id=scholar_profile_id) return success_payload( request, data={"message": "Scholar deleted."}, @@ -551,14 +499,7 @@ async def update_scholar_image_url( message=str(exc), ) from exc - logger.info( - "api.scholars.image_url_updated", - extra={ - "event": "api.scholars.image_url_updated", - "user_id": current_user.id, - "scholar_profile_id": updated.id, - }, - ) + structured_log(logger, "info", "api.scholars.image_url_updated", user_id=current_user.id, scholar_profile_id=updated.id) return success_payload( request, data=_serialize_scholar(updated), @@ -600,16 +541,7 @@ async def upload_scholar_image( ) from exc image_size = len(image_bytes) - logger.info( - "api.scholars.image_uploaded", - extra={ - "event": "api.scholars.image_uploaded", - "user_id": current_user.id, - "scholar_profile_id": updated.id, - "content_type": image.content_type, - "size_bytes": image_size, - }, - ) + structured_log(logger, "info", "api.scholars.image_uploaded", user_id=current_user.id, scholar_profile_id=updated.id, content_type=image.content_type, size_bytes=image_size) return success_payload( request, data=_serialize_scholar(updated), @@ -643,14 +575,7 @@ async def clear_scholar_image_customization( profile=profile, upload_dir=settings.scholar_image_upload_dir, ) - logger.info( - "api.scholars.image_customization_cleared", - extra={ - "event": "api.scholars.image_customization_cleared", - "user_id": current_user.id, - "scholar_profile_id": updated.id, - }, - ) + structured_log(logger, "info", "api.scholars.image_customization_cleared", user_id=current_user.id, scholar_profile_id=updated.id) return success_payload( request, data=_serialize_scholar(updated), diff --git a/app/api/routers/settings.py b/app/api/routers/settings.py index 5a2ef91..e0c3f26 100644 --- a/app/api/routers/settings.py +++ b/app/api/routers/settings.py @@ -11,6 +11,7 @@ from app.api.responses import success_payload from app.api.schemas import SettingsEnvelope, SettingsUpdateRequest from app.db.models import User from app.db.session import get_db_session +from app.logging_utils import structured_log from app.services.domains.ingestion import safety as run_safety_service from app.services.domains.settings import application as user_settings_service from app.settings import settings as settings_module @@ -83,16 +84,11 @@ async def _clear_expired_cooldown_with_log( return await db_session.commit() await db_session.refresh(user_settings) - logger.info( - "api.settings.safety_cooldown_cleared", - extra={ - "event": "api.settings.safety_cooldown_cleared", - "user_id": user_id, - "reason": previous_safety_state.get("cooldown_reason"), - "cooldown_until": previous_safety_state.get("cooldown_until"), - "metric_name": "api_settings_safety_cooldown_cleared_total", - "metric_value": 1, - }, + structured_log( + logger, "info", "api.settings.safety_cooldown_cleared", + user_id=user_id, + reason=previous_safety_state.get("cooldown_reason"), + cooldown_until=previous_safety_state.get("cooldown_until"), ) @@ -100,21 +96,6 @@ def _effective_auto_run_enabled(payload: SettingsUpdateRequest) -> bool: return bool(payload.auto_run_enabled) and settings_module.ingestion_automation_allowed -def _log_settings_update(*, user_id: int, updated) -> None: - logger.info( - "api.settings.updated", - extra={ - "event": "api.settings.updated", - "user_id": user_id, - "auto_run_enabled": updated.auto_run_enabled, - "run_interval_minutes": updated.run_interval_minutes, - "request_delay_seconds": updated.request_delay_seconds, - "nav_visible_pages": updated.nav_visible_pages, - "openalex_api_key": "SET" if updated.openalex_api_key else "UNSET", - }, - ) - - @router.get( "", response_model=SettingsEnvelope, @@ -182,7 +163,15 @@ async def update_settings( user_id=current_user.id, user_settings=updated, ) - _log_settings_update(user_id=current_user.id, updated=updated) + structured_log( + logger, "info", "api.settings.updated", + user_id=current_user.id, + auto_run_enabled=updated.auto_run_enabled, + run_interval_minutes=updated.run_interval_minutes, + request_delay_seconds=updated.request_delay_seconds, + nav_visible_pages=updated.nav_visible_pages, + openalex_api_key="SET" if updated.openalex_api_key else "UNSET", + ) return success_payload( request, data=_serialize_settings(updated), diff --git a/app/auth/runtime.py b/app/auth/runtime.py index 777e799..c379cd3 100644 --- a/app/auth/runtime.py +++ b/app/auth/runtime.py @@ -6,6 +6,7 @@ from fastapi import Request from sqlalchemy.ext.asyncio import AsyncSession from app.auth.session import clear_session_user, get_session_user, set_session_user +from app.logging_utils import structured_log from app.db.models import User from app.security.csrf import CSRF_SESSION_KEY from app.services.domains.users import application as user_service @@ -28,13 +29,7 @@ async def get_authenticated_user( user = await user_service.get_user_by_id(db_session, session_user.id) if user is None or not user.is_active: - logger.info( - "auth.session_invalidated", - extra={ - "event": "auth.session_invalidated", - "session_user_id": session_user.id, - }, - ) + structured_log(logger, "info", "auth.session_invalidated", session_user_id=session_user.id) invalidate_session(request) return None diff --git a/app/db/session.py b/app/db/session.py index c05266e..2223216 100644 --- a/app/db/session.py +++ b/app/db/session.py @@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import ( ) from sqlalchemy.pool import NullPool +from app.logging_utils import structured_log from app.settings import settings logger = logging.getLogger(__name__) @@ -30,13 +31,10 @@ def _normalized_pool_mode(raw_mode: str) -> str: return "queue" if mode in {"null", "queue"}: return mode - logger.warning( - "db.invalid_pool_mode_fallback", - extra={ - "event": "db.invalid_pool_mode_fallback", - "database_pool_mode": raw_mode, - "fallback_mode": "queue", - }, + structured_log( + logger, "warning", "db.invalid_pool_mode_fallback", + database_pool_mode=raw_mode, + fallback_mode="queue", ) return "queue" @@ -56,12 +54,9 @@ def get_engine() -> AsyncEngine: engine_kwargs["pool_timeout"] = max(1, int(settings.database_pool_timeout_seconds)) _engine = create_async_engine(settings.database_url, **engine_kwargs) - logger.info( - "db.engine_initialized", - extra={ - "event": "db.engine_initialized", - "pool_mode": pool_mode, - }, + structured_log( + logger, "info", "db.engine_initialized", + pool_mode=pool_mode, ) return _engine @@ -86,7 +81,7 @@ async def check_database() -> bool: result = await conn.execute(text("SELECT 1")) return result.scalar_one() == 1 except Exception: - logger.exception("db.healthcheck_failed", extra={"event": "db.healthcheck_failed"}) + logger.exception("db.healthcheck_failed") return False @@ -94,6 +89,6 @@ async def close_engine() -> None: global _engine, _session_factory if _engine is not None: await _engine.dispose() - logger.info("db.engine_disposed", extra={"event": "db.engine_disposed"}) + structured_log(logger, "info", "db.engine_disposed") _engine = None _session_factory = None diff --git a/app/http/middleware.py b/app/http/middleware.py index ebe72f8..1cbea46 100644 --- a/app/http/middleware.py +++ b/app/http/middleware.py @@ -9,6 +9,7 @@ from starlette.requests import Request from starlette.responses import Response from app.logging_context import set_request_id +from app.logging_utils import structured_log REQUEST_ID_HEADER = "X-Request-ID" DEFAULT_PERMISSIONS_POLICY = ( @@ -61,13 +62,10 @@ class RequestLoggingMiddleware(BaseHTTPMiddleware): start = time.perf_counter() should_log = self._log_requests and not self._is_skipped_path(request.url.path) if should_log: - logger.debug( - "request.started", - extra={ - "event": "request.started", - "method": request.method, - "path": request.url.path, - }, + structured_log( + logger, "debug", "request.started", + method=request.method, + path=request.url.path, ) try: @@ -77,7 +75,6 @@ class RequestLoggingMiddleware(BaseHTTPMiddleware): logger.exception( "request.failed", extra={ - "event": "request.failed", "method": request.method, "path": request.url.path, "duration_ms": duration_ms, @@ -88,15 +85,12 @@ class RequestLoggingMiddleware(BaseHTTPMiddleware): duration_ms = int((time.perf_counter() - start) * 1000) response.headers[REQUEST_ID_HEADER] = request_id if should_log: - logger.debug( - "request.completed", - extra={ - "event": "request.completed", - "method": request.method, - "path": request.url.path, - "status_code": response.status_code, - "duration_ms": duration_ms, - }, + structured_log( + logger, "debug", "request.completed", + method=request.method, + path=request.url.path, + status_code=response.status_code, + duration_ms=duration_ms, ) return response finally: diff --git a/app/main.py b/app/main.py index 56ddf58..522a552 100644 --- a/app/main.py +++ b/app/main.py @@ -21,6 +21,7 @@ from app.http.middleware import ( parse_skip_paths, ) from app.logging_config import configure_logging, parse_redact_fields +from app.logging_utils import structured_log from app.security.csrf import CSRFMiddleware from app.services.domains.ingestion.scheduler import SchedulerService from app.settings import settings @@ -50,22 +51,15 @@ scheduler_service = SchedulerService( ) -def _log_startup_build_marker() -> None: - logger.info( - "app.startup_build_marker", - extra={ - "event": "app.startup_build_marker", - "build_marker": BUILD_MARKER, - "frontend_enabled": settings.frontend_enabled, - "scheduler_enabled": settings.scheduler_enabled, - "log_format": settings.log_format, - }, - ) - - @asynccontextmanager async def lifespan(_: FastAPI): - _log_startup_build_marker() + structured_log( + logger, "info", "app.startup_build_marker", + build_marker=BUILD_MARKER, + frontend_enabled=settings.frontend_enabled, + scheduler_enabled=settings.scheduler_enabled, + log_format=settings.log_format, + ) from app.db.session import get_session_factory from sqlalchemy import text @@ -74,7 +68,7 @@ async def lifespan(_: FastAPI): async with session_factory() as session: await session.execute(text("UPDATE crawl_runs SET status = 'failed' WHERE status::text IN ('running', 'resolving')")) await session.commit() - logger.info("app.startup_orphaned_runs_cleaned", extra={"event": "app.startup_orphaned_runs_cleaned"}) + structured_log(logger, "info", "app.startup_orphaned_runs_cleaned") except Exception as e: logger.error(f"Failed to clean orphaned runs at startup: {e}") diff --git a/app/security/csrf.py b/app/security/csrf.py index 70eecac..d2ed1c5 100644 --- a/app/security/csrf.py +++ b/app/security/csrf.py @@ -9,6 +9,7 @@ from starlette.requests import Request from starlette.responses import JSONResponse, PlainTextResponse, Response from starlette.types import Message +from app.logging_utils import structured_log CSRF_SESSION_KEY = "csrf_token" CSRF_FORM_FIELD = "csrf_token" @@ -36,13 +37,10 @@ class CSRFMiddleware(BaseHTTPMiddleware): session_token = request.session.get(CSRF_SESSION_KEY) if not session_token: - logger.warning( - "csrf.missing_session_token", - extra={ - "event": "csrf.missing_session_token", - "method": request.method, - "path": request.url.path, - }, + structured_log( + logger, "warning", "csrf.missing_session_token", + method=request.method, + path=request.url.path, ) return self._csrf_error_response( request, @@ -57,13 +55,10 @@ class CSRFMiddleware(BaseHTTPMiddleware): request._receive = self._build_receive(body) # type: ignore[attr-defined] if not request_token or not compare_digest(str(session_token), str(request_token)): - logger.warning( - "csrf.invalid_token", - extra={ - "event": "csrf.invalid_token", - "method": request.method, - "path": request.url.path, - }, + structured_log( + logger, "warning", "csrf.invalid_token", + method=request.method, + path=request.url.path, ) return self._csrf_error_response( request, diff --git a/app/services/domains/arxiv/client.py b/app/services/domains/arxiv/client.py index 278ca0f..7bbf87c 100644 --- a/app/services/domains/arxiv/client.py +++ b/app/services/domains/arxiv/client.py @@ -5,6 +5,7 @@ import logging import httpx +from app.logging_utils import structured_log from app.services.domains.arxiv.cache import ( build_query_fingerprint, get_cached_feed, @@ -102,17 +103,9 @@ class ArxivClient: if self._cache_enabled: cached = await get_cached_feed(query_fingerprint=query_fingerprint) if cached is not None: - _log_cache_event( - event_name="arxiv.cache_hit", - query_fingerprint=query_fingerprint, - source_path=source_path, - ) + structured_log(logger, "info", "arxiv.cache_hit", query_fingerprint=query_fingerprint, source_path=source_path) return cached - _log_cache_event( - event_name="arxiv.cache_miss", - query_fingerprint=query_fingerprint, - source_path=source_path, - ) + structured_log(logger, "info", "arxiv.cache_miss", query_fingerprint=query_fingerprint, source_path=source_path) return await run_with_inflight_dedupe( query_fingerprint=query_fingerprint, fetch_feed=lambda: self._fetch_live_feed( @@ -222,9 +215,10 @@ async def _request_arxiv_feed( source_path = _source_path_from_params(params) cooldown_status = await get_arxiv_cooldown_status() if cooldown_status.is_active: - _log_request_skipped_for_cooldown( + structured_log( + logger, "warning", "arxiv.request_skipped_cooldown", source_path=source_path, - cooldown_remaining_seconds=cooldown_status.remaining_seconds, + cooldown_remaining_seconds=float(cooldown_status.remaining_seconds), ) raise ArxivRateLimitError( f"arXiv global cooldown active ({cooldown_status.remaining_seconds:.0f}s remaining)" @@ -280,32 +274,3 @@ def _source_path_from_params(params: dict[str, object]) -> str: return ARXIV_SOURCE_PATH_UNKNOWN -def _log_cache_event( - *, - event_name: str, - query_fingerprint: str, - source_path: str, -) -> None: - logger.info( - event_name, - extra={ - "event": event_name, - "query_fingerprint": query_fingerprint, - "source_path": source_path, - }, - ) - - -def _log_request_skipped_for_cooldown( - *, - source_path: str, - cooldown_remaining_seconds: float, -) -> None: - logger.warning( - "arxiv.request_skipped_cooldown", - extra={ - "event": "arxiv.request_skipped_cooldown", - "source_path": source_path, - "cooldown_remaining_seconds": float(cooldown_remaining_seconds), - }, - ) diff --git a/app/services/domains/arxiv/gateway.py b/app/services/domains/arxiv/gateway.py index 6782025..67d9543 100644 --- a/app/services/domains/arxiv/gateway.py +++ b/app/services/domains/arxiv/gateway.py @@ -5,6 +5,7 @@ import re from typing import TYPE_CHECKING, Protocol import unicodedata +from app.logging_utils import structured_log from app.services.domains.arxiv.client import ArxivClient from app.services.domains.arxiv.errors import ArxivRateLimitError from app.services.domains.arxiv.types import ArxivFeed @@ -112,7 +113,7 @@ class HttpArxivGateway: except ArxivRateLimitError: raise except Exception as exc: - logger.debug("arxiv.query_failed", extra={"event": "arxiv.query_failed", "error": str(exc)}) + structured_log(logger, "debug", "arxiv.query_failed", error=str(exc)) return None @@ -136,6 +137,6 @@ def _author_surname(scholar_label: str | None) -> str | None: def _first_discovered_id(result: ArxivFeed) -> str | None: for entry in result.entries: if entry.arxiv_id: - logger.debug("arxiv.id_discovered", extra={"event": "arxiv.id_discovered", "arxiv_id": entry.arxiv_id}) + structured_log(logger, "debug", "arxiv.id_discovered", arxiv_id=entry.arxiv_id) return entry.arxiv_id return None diff --git a/app/services/domains/arxiv/rate_limit.py b/app/services/domains/arxiv/rate_limit.py index 6506035..52e2c3f 100644 --- a/app/services/domains/arxiv/rate_limit.py +++ b/app/services/domains/arxiv/rate_limit.py @@ -12,6 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import ArxivRuntimeState from app.db.session import get_session_factory +from app.logging_utils import structured_log from app.services.domains.arxiv.constants import ( ARXIV_RATE_LIMIT_LOCK_KEY, ARXIV_RATE_LIMIT_LOCK_NAMESPACE, @@ -83,14 +84,12 @@ async def _run_serialized_fetch( response_status=int(response.status_code), source_path=source_path, ) - _log_request_completed( - response_status=int(response.status_code), + structured_log( + logger, "info", "arxiv.request_completed", + status_code=int(response.status_code), wait_seconds=wait_seconds, + cooldown_remaining_seconds=_cooldown_remaining_seconds(runtime_state.cooldown_until, now_utc=datetime.now(timezone.utc)), source_path=source_path, - cooldown_remaining_seconds=_cooldown_remaining_seconds( - runtime_state.cooldown_until, - now_utc=datetime.now(timezone.utc), - ), ) return response, hit_rate_limit @@ -128,18 +127,10 @@ async def _wait_for_allowed_slot_or_raise( now_utc = datetime.now(timezone.utc) cooldown_seconds = _cooldown_remaining_seconds(runtime_state.cooldown_until, now_utc=now_utc) if cooldown_seconds > 0: - _log_request_scheduled( - wait_seconds=0.0, - source_path=source_path, - cooldown_remaining_seconds=cooldown_seconds, - ) + structured_log(logger, "info", "arxiv.request_scheduled", wait_seconds=0.0, source_path=source_path, cooldown_remaining_seconds=cooldown_seconds) raise ArxivRateLimitError(f"arXiv global cooldown active ({cooldown_seconds:.0f}s remaining)") wait_seconds = _next_allowed_wait_seconds(runtime_state.next_allowed_at, now_utc=now_utc) - _log_request_scheduled( - wait_seconds=wait_seconds, - source_path=source_path, - cooldown_remaining_seconds=0.0, - ) + structured_log(logger, "info", "arxiv.request_scheduled", wait_seconds=wait_seconds, source_path=source_path, cooldown_remaining_seconds=0.0) if wait_seconds > 0: await asyncio.sleep(wait_seconds) return wait_seconds @@ -156,9 +147,10 @@ def _record_post_response_state( if response_status == 429: cooldown_seconds = _cooldown_seconds() runtime_state.cooldown_until = now_utc + timedelta(seconds=cooldown_seconds) - _log_cooldown_activated( - source_path=source_path, + structured_log( + logger, "warning", "arxiv.cooldown_activated", cooldown_remaining_seconds=cooldown_seconds, + source_path=source_path, ) return True if _cooldown_remaining_seconds(runtime_state.cooldown_until, now_utc=now_utc) <= 0: @@ -196,52 +188,3 @@ def _cooldown_seconds() -> float: return max(float(settings.arxiv_rate_limit_cooldown_seconds), 0.0) -def _log_request_scheduled( - *, - wait_seconds: float, - source_path: str, - cooldown_remaining_seconds: float, -) -> None: - logger.info( - "arxiv.request_scheduled", - extra={ - "event": "arxiv.request_scheduled", - "wait_seconds": float(wait_seconds), - "cooldown_remaining_seconds": float(cooldown_remaining_seconds), - "source_path": source_path, - }, - ) - - -def _log_request_completed( - *, - response_status: int, - wait_seconds: float, - source_path: str, - cooldown_remaining_seconds: float, -) -> None: - logger.info( - "arxiv.request_completed", - extra={ - "event": "arxiv.request_completed", - "status_code": int(response_status), - "wait_seconds": float(wait_seconds), - "cooldown_remaining_seconds": float(cooldown_remaining_seconds), - "source_path": source_path, - }, - ) - - -def _log_cooldown_activated( - *, - source_path: str, - cooldown_remaining_seconds: float, -) -> None: - logger.warning( - "arxiv.cooldown_activated", - extra={ - "event": "arxiv.cooldown_activated", - "cooldown_remaining_seconds": float(cooldown_remaining_seconds), - "source_path": source_path, - }, - ) diff --git a/app/services/domains/crossref/application.py b/app/services/domains/crossref/application.py index f0e1311..7212101 100644 --- a/app/services/domains/crossref/application.py +++ b/app/services/domains/crossref/application.py @@ -10,6 +10,7 @@ from typing import TYPE_CHECKING from crossref.restful import Etiquette, Works from app.services.domains.doi.normalize import normalize_doi +from app.logging_utils import structured_log from app.settings import settings if TYPE_CHECKING: @@ -377,6 +378,6 @@ async def discover_doi_for_publication( author_surname=author_surname, ) if doi: - logger.debug("crossref.doi_discovered", extra={"event": "crossref.doi_discovered"}) + structured_log(logger, "debug", "crossref.doi_discovered") return doi return None diff --git a/app/services/domains/ingestion/scheduler.py b/app/services/domains/ingestion/scheduler.py index b05e84d..1f2267a 100644 --- a/app/services/domains/ingestion/scheduler.py +++ b/app/services/domains/ingestion/scheduler.py @@ -23,6 +23,7 @@ from app.services.domains.ingestion.application import ( RunBlockedBySafetyPolicyError, ScholarIngestionService, ) +from app.logging_utils import structured_log from app.services.domains.settings import application as user_settings_service from app.services.domains.scholar.source import LiveScholarSource from app.settings import settings @@ -89,31 +90,23 @@ class SchedulerService: async def start(self) -> None: if not self._enabled: - logger.info( - "scheduler.disabled", - extra={ - "event": "scheduler.disabled", - }, - ) + structured_log(logger, "info", "scheduler.disabled") return if self._task is not None: return self._task = asyncio.create_task(self._run_loop(), name="scholarr-scheduler") - logger.info( - "scheduler.started", - extra={ - "event": "scheduler.started", - "tick_seconds": self._tick_seconds, - "network_error_retries": self._network_error_retries, - "retry_backoff_seconds": self._retry_backoff_seconds, - "max_pages_per_scholar": self._max_pages_per_scholar, - "page_size": self._page_size, - "continuation_queue_enabled": self._continuation_queue_enabled, - "continuation_base_delay_seconds": self._continuation_base_delay_seconds, - "continuation_max_delay_seconds": self._continuation_max_delay_seconds, - "continuation_max_attempts": self._continuation_max_attempts, - "queue_batch_size": self._queue_batch_size, - }, + structured_log( + logger, "info", "scheduler.started", + tick_seconds=self._tick_seconds, + network_error_retries=self._network_error_retries, + retry_backoff_seconds=self._retry_backoff_seconds, + max_pages_per_scholar=self._max_pages_per_scholar, + page_size=self._page_size, + continuation_queue_enabled=self._continuation_queue_enabled, + continuation_base_delay_seconds=self._continuation_base_delay_seconds, + continuation_max_delay_seconds=self._continuation_max_delay_seconds, + continuation_max_attempts=self._continuation_max_attempts, + queue_batch_size=self._queue_batch_size, ) async def stop(self) -> None: @@ -126,7 +119,7 @@ class SchedulerService: pass finally: self._task = None - logger.info("scheduler.stopped", extra={"event": "scheduler.stopped"}) + structured_log(logger, "info", "scheduler.stopped") async def _run_loop(self) -> None: while True: @@ -137,9 +130,7 @@ class SchedulerService: except Exception: logger.exception( "scheduler.tick_failed", - extra={ - "event": "scheduler.tick_failed", - }, + extra={}, ) await asyncio.sleep(float(self._tick_seconds)) @@ -181,17 +172,12 @@ class SchedulerService: if cooldown_until is not None and cooldown_until.tzinfo is None: cooldown_until = cooldown_until.replace(tzinfo=timezone.utc) if cooldown_until is not None and cooldown_until > now_utc: - logger.info( - "scheduler.run_skipped_safety_cooldown_precheck", - extra={ - "event": "scheduler.run_skipped_safety_cooldown_precheck", - "user_id": int(user_id), - "reason": cooldown_reason, - "cooldown_until": cooldown_until, - "cooldown_remaining_seconds": int((cooldown_until - now_utc).total_seconds()), - "metric_name": "scheduler_run_skipped_safety_cooldown_total", - "metric_value": 1, - }, + structured_log( + logger, "info", "scheduler.run_skipped_safety_cooldown_precheck", + user_id=int(user_id), + reason=cooldown_reason, + cooldown_until=cooldown_until, + cooldown_remaining_seconds=int((cooldown_until - now_utc).total_seconds()), ) return None return _AutoRunCandidate( @@ -261,42 +247,34 @@ class SchedulerService: ) except RunAlreadyInProgressError: await session.rollback() - logger.info("scheduler.run_skipped_locked", extra={"event": "scheduler.run_skipped_locked", "user_id": candidate.user_id}) + structured_log(logger, "info", "scheduler.run_skipped_locked", user_id=candidate.user_id) return None except RunBlockedBySafetyPolicyError as exc: await session.rollback() - logger.info( - "scheduler.run_skipped_safety_cooldown", - extra={ - "event": "scheduler.run_skipped_safety_cooldown", - "user_id": candidate.user_id, - "reason": exc.safety_state.get("cooldown_reason"), - "cooldown_until": exc.safety_state.get("cooldown_until"), - "cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"), - "metric_name": "scheduler_run_skipped_safety_cooldown_total", - "metric_value": 1, - }, + structured_log( + logger, "info", "scheduler.run_skipped_safety_cooldown", + user_id=candidate.user_id, + reason=exc.safety_state.get("cooldown_reason"), + cooldown_until=exc.safety_state.get("cooldown_until"), + cooldown_remaining_seconds=exc.safety_state.get("cooldown_remaining_seconds"), ) return None except Exception: await session.rollback() - logger.exception("scheduler.run_failed", extra={"event": "scheduler.run_failed", "user_id": candidate.user_id}) + logger.exception("scheduler.run_failed", extra={"user_id": candidate.user_id}) return None async def _run_candidate(self, candidate: _AutoRunCandidate) -> None: run_summary = await self._run_candidate_ingestion(candidate=candidate) if run_summary is None: return - logger.info( - "scheduler.run_completed", - extra={ - "event": "scheduler.run_completed", - "user_id": candidate.user_id, - "run_id": run_summary.crawl_run_id, - "status": run_summary.status.value, - "scholar_count": run_summary.scholar_count, - "new_publication_count": run_summary.new_publication_count, - }, + structured_log( + logger, "info", "scheduler.run_completed", + user_id=candidate.user_id, + run_id=run_summary.crawl_run_id, + status=run_summary.status.value, + scholar_count=run_summary.scholar_count, + new_publication_count=run_summary.new_publication_count, ) async def _drain_continuation_queue(self) -> None: @@ -326,16 +304,13 @@ class SchedulerService: ) await session.commit() if dropped is not None: - logger.warning( - "scheduler.queue_item_dropped_max_attempts", - extra={ - "event": "scheduler.queue_item_dropped_max_attempts", - "queue_item_id": job.id, - "user_id": job.user_id, - "scholar_profile_id": job.scholar_profile_id, - "attempt_count": job.attempt_count, - "max_attempts": self._continuation_max_attempts, - }, + structured_log( + logger, "warning", "scheduler.queue_item_dropped_max_attempts", + queue_item_id=job.id, + user_id=job.user_id, + scholar_profile_id=job.scholar_profile_id, + attempt_count=job.attempt_count, + max_attempts=self._continuation_max_attempts, ) return True @@ -376,14 +351,11 @@ class SchedulerService: ) await session.commit() if dropped is not None: - logger.info( - "scheduler.queue_item_dropped_scholar_unavailable", - extra={ - "event": "scheduler.queue_item_dropped_scholar_unavailable", - "queue_item_id": job.id, - "user_id": job.user_id, - "scholar_profile_id": job.scholar_profile_id, - }, + structured_log( + logger, "info", "scheduler.queue_item_dropped_scholar_unavailable", + queue_item_id=job.id, + user_id=job.user_id, + scholar_profile_id=job.scholar_profile_id, ) return False @@ -398,9 +370,9 @@ class SchedulerService: error="run_already_in_progress", ) await recovery_session.commit() - logger.info( - "scheduler.queue_item_deferred_lock", - extra={"event": "scheduler.queue_item_deferred_lock", "queue_item_id": job.id, "user_id": job.user_id}, + structured_log( + logger, "info", "scheduler.queue_item_deferred_lock", + queue_item_id=job.id, user_id=job.user_id, ) async def _reschedule_queue_job_safety_cooldown( @@ -422,17 +394,12 @@ class SchedulerService: error=str(exc.message), ) await recovery_session.commit() - logger.info( - "scheduler.queue_item_deferred_safety_cooldown", - extra={ - "event": "scheduler.queue_item_deferred_safety_cooldown", - "queue_item_id": job.id, - "user_id": job.user_id, - "reason": exc.safety_state.get("cooldown_reason"), - "cooldown_remaining_seconds": cooldown_remaining_seconds, - "metric_name": "scheduler_queue_item_deferred_safety_cooldown_total", - "metric_value": 1, - }, + structured_log( + logger, "info", "scheduler.queue_item_deferred_safety_cooldown", + queue_item_id=job.id, + user_id=job.user_id, + reason=exc.safety_state.get("cooldown_reason"), + cooldown_remaining_seconds=cooldown_remaining_seconds, ) async def _reschedule_queue_job_after_exception( @@ -455,9 +422,9 @@ class SchedulerService: error=str(exc), ) await recovery_session.commit() - logger.warning( - "scheduler.queue_item_dropped_after_exception", - extra={"event": "scheduler.queue_item_dropped_after_exception", "queue_item_id": job.id, "user_id": job.user_id, "attempt_count": queue_item.attempt_count}, + structured_log( + logger, "warning", "scheduler.queue_item_dropped_after_exception", + queue_item_id=job.id, user_id=job.user_id, attempt_count=queue_item.attempt_count, ) return delay_seconds = queue_service.compute_backoff_seconds( @@ -473,7 +440,7 @@ class SchedulerService: error=str(exc), ) await recovery_session.commit() - logger.exception("scheduler.queue_item_run_failed", extra={"event": "scheduler.queue_item_run_failed", "queue_item_id": job.id, "user_id": job.user_id}) + logger.exception("scheduler.queue_item_run_failed", extra={"queue_item_id": job.id, "user_id": job.user_id}) async def _run_ingestion_for_queue_job( self, @@ -514,28 +481,6 @@ class SchedulerService: await self._reschedule_queue_job_after_exception(job, exc=exc) return None - @staticmethod - def _log_queue_item_resolved( - *, - event_name: str, - job: queue_service.ContinuationQueueJob, - run_summary, - attempt_count: int | None = None, - delay_seconds: int | None = None, - ) -> None: - payload = { - "event": event_name, - "queue_item_id": job.id, - "user_id": job.user_id, - "run_id": run_summary.crawl_run_id, - "status": run_summary.status.value, - } - if attempt_count is not None: - payload["attempt_count"] = attempt_count - if delay_seconds is not None: - payload["delay_seconds"] = delay_seconds - logger.info(event_name, extra=payload) - async def _finalize_queue_job_after_run(self, job: queue_service.ContinuationQueueJob, run_summary) -> None: session_factory = get_session_factory() async with session_factory() as session: @@ -543,35 +488,45 @@ class SchedulerService: queue_item = await queue_service.reset_attempt_count(session, job_id=job.id) await session.commit() if queue_item is None: - self._log_queue_item_resolved(event_name="scheduler.queue_item_resolved", job=job, run_summary=run_summary) + structured_log( + logger, "info", "scheduler.queue_item_resolved", + queue_item_id=job.id, user_id=job.user_id, + run_id=run_summary.crawl_run_id, status=run_summary.status.value, + ) return - self._log_queue_item_resolved( - event_name="scheduler.queue_item_progressed", - job=job, - run_summary=run_summary, + structured_log( + logger, "info", "scheduler.queue_item_progressed", + queue_item_id=job.id, user_id=job.user_id, + run_id=run_summary.crawl_run_id, status=run_summary.status.value, attempt_count=int(queue_item.attempt_count), ) return queue_item = await queue_service.increment_attempt_count(session, job_id=job.id) if queue_item is None: await session.commit() - self._log_queue_item_resolved(event_name="scheduler.queue_item_resolved", job=job, run_summary=run_summary) + structured_log( + logger, "info", "scheduler.queue_item_resolved", + queue_item_id=job.id, user_id=job.user_id, + run_id=run_summary.crawl_run_id, status=run_summary.status.value, + ) return if int(queue_item.attempt_count) >= self._continuation_max_attempts: await queue_service.mark_dropped(session, job_id=job.id, reason="max_attempts_after_run") await session.commit() - logger.warning( - "scheduler.queue_item_dropped_max_attempts_after_run", - extra={"event": "scheduler.queue_item_dropped_max_attempts_after_run", "queue_item_id": job.id, "user_id": job.user_id, "attempt_count": queue_item.attempt_count, "run_id": run_summary.crawl_run_id, "status": run_summary.status.value}, + structured_log( + logger, "warning", "scheduler.queue_item_dropped_max_attempts_after_run", + queue_item_id=job.id, user_id=job.user_id, + attempt_count=queue_item.attempt_count, + run_id=run_summary.crawl_run_id, status=run_summary.status.value, ) return delay_seconds = queue_service.compute_backoff_seconds(base_seconds=self._continuation_base_delay_seconds, attempt_count=int(queue_item.attempt_count), max_seconds=self._continuation_max_delay_seconds) await queue_service.reschedule_job(session, job_id=job.id, delay_seconds=delay_seconds, reason=queue_item.reason, error=queue_item.last_error) await session.commit() - self._log_queue_item_resolved( - event_name="scheduler.queue_item_rescheduled_failed", - job=job, - run_summary=run_summary, + structured_log( + logger, "info", "scheduler.queue_item_rescheduled_failed", + queue_item_id=job.id, user_id=job.user_id, + run_id=run_summary.crawl_run_id, status=run_summary.status.value, attempt_count=int(queue_item.attempt_count), delay_seconds=delay_seconds, ) @@ -600,14 +555,12 @@ class SchedulerService: max_attempts=settings.pdf_auto_retry_max_attempts, ) if processed > 0: - logger.info("scheduler.pdf_queue_drain_completed", extra={ - "event": "scheduler.pdf_queue_drain_completed", - "processed_count": processed, - }) + structured_log( + logger, "info", "scheduler.pdf_queue_drain_completed", + processed_count=processed, + ) except Exception: - logger.exception("scheduler.pdf_queue_drain_failed", extra={ - "event": "scheduler.pdf_queue_drain_failed", - }) + logger.exception("scheduler.pdf_queue_drain_failed", extra={}) async def _load_request_delay_for_user( self, diff --git a/app/services/domains/openalex/client.py b/app/services/domains/openalex/client.py index a85b630..f6db5fd 100644 --- a/app/services/domains/openalex/client.py +++ b/app/services/domains/openalex/client.py @@ -84,7 +84,7 @@ class OpenAlexClient: ) raise OpenAlexRateLimitError("Rate limit exceeded fetching OpenAlex work by DOI") if response.status_code >= 400: - logger.warning("OpenAlex API error: %s %s", response.status_code, response.text) + logger.warning("OpenAlex API error: %s %s", response.status_code, response.text[:500]) raise OpenAlexClientError(f"API Error {response.status_code}") data = response.json() @@ -134,7 +134,7 @@ class OpenAlexClient: ) raise OpenAlexRateLimitError("Rate limit exceeded fetching OpenAlex works list") if response.status_code >= 400: - logger.warning("OpenAlex API error (filters=%s): %s %s", filters, response.status_code, response.text) + logger.warning("OpenAlex API error (filters=%s): %s %s", filters, response.status_code, response.text[:500]) raise OpenAlexClientError(f"API Error {response.status_code}") data = response.json() diff --git a/app/services/domains/publications/dedup.py b/app/services/domains/publications/dedup.py index 6615d3b..5bf1d43 100644 --- a/app/services/domains/publications/dedup.py +++ b/app/services/domains/publications/dedup.py @@ -10,6 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased from app.db.models import Publication, PublicationIdentifier, ScholarPublication +from app.logging_utils import structured_log from app.services.domains.ingestion.fingerprints import ( canonical_title_text_for_dedup, canonical_title_tokens_for_dedup, @@ -104,14 +105,7 @@ async def merge_duplicate_publication( await _migrate_scholar_links(db_session, winner_id=winner_id, dup_id=dup_id) await _migrate_identifiers(db_session, winner_id=winner_id, dup_id=dup_id) await db_session.execute(delete(Publication).where(Publication.id == dup_id)) - logger.info( - "publications.identifier_merge", - extra={ - "event": "publications.identifier_merge", - "winner_id": winner_id, - "dup_id": dup_id, - }, - ) + structured_log(logger, "info", "publications.identifier_merge", winner_id=winner_id, dup_id=dup_id) async def _load_publication( diff --git a/app/services/domains/publications/enrichment.py b/app/services/domains/publications/enrichment.py index 381107a..8a6b04f 100644 --- a/app/services/domains/publications/enrichment.py +++ b/app/services/domains/publications/enrichment.py @@ -9,6 +9,7 @@ from app.services.domains.publications.pdf_queue import ( enqueue_retry_pdf_job, overlay_pdf_job_state, ) +from app.logging_utils import structured_log from app.services.domains.publications.types import PublicationListItem logger = logging.getLogger(__name__) @@ -29,14 +30,7 @@ async def schedule_missing_pdf_enrichment_for_user( rows=items, max_items=max_items, ) - logger.info( - "publications.enrichment.scheduled", - extra={ - "event": "publications.enrichment.scheduled", - "user_id": user_id, - "publication_count": len(queued_ids), - }, - ) + structured_log(logger, "info", "publications.enrichment.scheduled", user_id=user_id, publication_count=len(queued_ids)) return len(queued_ids) @@ -53,15 +47,7 @@ async def schedule_retry_pdf_enrichment_for_row( request_email=request_email, row=item, ) - logger.info( - "publications.enrichment.retry_scheduled", - extra={ - "event": "publications.enrichment.retry_scheduled", - "user_id": user_id, - "publication_id": item.publication_id, - "queued": queued, - }, - ) + structured_log(logger, "info", "publications.enrichment.retry_scheduled", user_id=user_id, publication_id=item.publication_id, queued=queued) return queued diff --git a/app/services/domains/publications/pdf_queue.py b/app/services/domains/publications/pdf_queue.py index 3cb0de4..83c9031 100644 --- a/app/services/domains/publications/pdf_queue.py +++ b/app/services/domains/publications/pdf_queue.py @@ -27,6 +27,7 @@ from app.services.domains.unpaywall.application import ( FAILURE_RESOLUTION_EXCEPTION, OaResolutionOutcome, ) +from app.logging_utils import structured_log from app.settings import settings PDF_STATUS_UNTRACKED = "untracked" @@ -320,10 +321,7 @@ def _drop_finished_task(task: asyncio.Task[None]) -> None: try: task.result() except Exception: - logger.exception( - "publications.pdf_queue.task_failed", - extra={"event": "publications.pdf_queue.task_failed"}, - ) + logger.exception("publications.pdf_queue.task_failed") async def _mark_attempt_started( @@ -474,14 +472,7 @@ async def _resolve_publication_row( # Propagate upward so the batch loop can stop immediately. raise except Exception as exc: # pragma: no cover - defensive network boundary - logger.warning( - "publications.pdf_queue.resolve_failed", - extra={ - "event": "publications.pdf_queue.resolve_failed", - "publication_id": row.publication_id, - "error": str(exc), - }, - ) + structured_log(logger, "warning", "publications.pdf_queue.resolve_failed", publication_id=row.publication_id, error=str(exc)) outcome = _failed_outcome(row=row) arxiv_rate_limited = False await _persist_outcome( @@ -523,19 +514,9 @@ async def _run_resolution_task( ) if arxiv_rate_limited and arxiv_lookup_allowed: arxiv_lookup_allowed = False - logger.warning( - "publications.pdf_queue.arxiv_batch_disabled", - extra={ - "event": "publications.pdf_queue.arxiv_batch_disabled", - "detail": "arXiv temporarily disabled for remaining batch after rate limit", - }, - ) + structured_log(logger, "warning", "publications.pdf_queue.arxiv_batch_disabled", detail="arXiv temporarily disabled for remaining batch after rate limit") except OpenAlexBudgetExhaustedError: - logger.warning( - "publications.pdf_queue.budget_exhausted", - extra={"event": "publications.pdf_queue.budget_exhausted", - "detail": "Stopping PDF resolution batch — OpenAlex daily budget exhausted"}, - ) + structured_log(logger, "warning", "publications.pdf_queue.budget_exhausted", detail="Stopping PDF resolution batch — OpenAlex daily budget exhausted") break diff --git a/app/services/domains/publications/pdf_resolution_pipeline.py b/app/services/domains/publications/pdf_resolution_pipeline.py index 7ad1f8d..f17d65b 100644 --- a/app/services/domains/publications/pdf_resolution_pipeline.py +++ b/app/services/domains/publications/pdf_resolution_pipeline.py @@ -9,6 +9,7 @@ from app.services.domains.arxiv.guards import arxiv_skip_reason_for_item from app.services.domains.openalex.client import OpenAlexBudgetExhaustedError from app.services.domains.publications.types import PublicationListItem from app.services.domains.unpaywall.application import OaResolutionOutcome, resolve_publication_oa_outcomes +from app.logging_utils import structured_log from app.settings import settings logger = logging.getLogger(__name__) @@ -44,13 +45,7 @@ async def resolve_publication_pdf_outcome_for_row( except ArxivRateLimitError: arxiv_rate_limited = True arxiv_outcome = None - logger.warning( - "publications.pdf_resolution.arxiv_rate_limited", - extra={ - "event": "publications.pdf_resolution.arxiv_rate_limited", - "publication_id": int(row.publication_id), - }, - ) + structured_log(logger, "warning", "publications.pdf_resolution.arxiv_rate_limited", publication_id=int(row.publication_id)) if arxiv_outcome and arxiv_outcome.pdf_url: return PipelineOutcome(arxiv_outcome, None, arxiv_rate_limited=arxiv_rate_limited) @@ -99,10 +94,7 @@ async def _openalex_outcome( # Re-raise so the caller's batch loop can stop hitting the API. raise except Exception as exc: - logger.warning( - "publications.pdf_resolution.openalex_failed", - extra={"event": "publications.pdf_resolution.openalex_failed", "error": str(exc)}, - ) + structured_log(logger, "warning", "publications.pdf_resolution.openalex_failed", error=str(exc)) return None @@ -115,12 +107,12 @@ async def _arxiv_outcome( from app.services.domains.arxiv.application import discover_arxiv_id_for_publication if not allow_lookup: - _log_arxiv_skip(row=row, skip_reason="batch_arxiv_cooldown_active") + structured_log(logger, "info", "publications.pdf_resolution.arxiv_skipped", publication_id=int(row.publication_id), skip_reason="batch_arxiv_cooldown_active") return None skip_reason = arxiv_skip_reason_for_item(item=row) if skip_reason is not None: - _log_arxiv_skip(row=row, skip_reason=skip_reason) + structured_log(logger, "info", "publications.pdf_resolution.arxiv_skipped", publication_id=int(row.publication_id), skip_reason=skip_reason) return None try: @@ -138,23 +130,10 @@ async def _arxiv_outcome( except ArxivRateLimitError: raise # propagate so orchestration can switch to non-arXiv fallback except Exception as exc: - logger.warning( - "publications.pdf_resolution.arxiv_failed", - extra={"event": "publications.pdf_resolution.arxiv_failed", "error": str(exc)}, - ) + structured_log(logger, "warning", "publications.pdf_resolution.arxiv_failed", error=str(exc)) return None -def _log_arxiv_skip(*, row: PublicationListItem, skip_reason: str) -> None: - logger.info( - "publications.pdf_resolution.arxiv_skipped", - extra={ - "event": "publications.pdf_resolution.arxiv_skipped", - "publication_id": int(row.publication_id), - "skip_reason": skip_reason, - }, - ) - async def _oa_outcome( *, diff --git a/app/services/domains/scholar/source.py b/app/services/domains/scholar/source.py index cf2ea0a..50d27ca 100644 --- a/app/services/domains/scholar/source.py +++ b/app/services/domains/scholar/source.py @@ -9,6 +9,7 @@ from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen +from app.logging_utils import structured_log from app.services.domains.scholar import rate_limit as scholar_rate_limit from app.settings import settings @@ -100,15 +101,12 @@ class LiveScholarSource: cstart=cstart, pagesize=pagesize, ) - logger.debug( - "scholar_source.fetch_started", - extra={ - "event": "scholar_source.fetch_started", - "scholar_id": scholar_id, - "requested_url": requested_url, - "cstart": cstart, - "pagesize": pagesize, - }, + structured_log( + logger, "debug", "scholar_source.fetch_started", + scholar_id=scholar_id, + requested_url=requested_url, + cstart=cstart, + pagesize=pagesize, ) return await self._fetch_with_global_throttle(requested_url) @@ -122,24 +120,18 @@ class LiveScholarSource: query=query, start=start, ) - logger.debug( - "scholar_source.search_fetch_started", - extra={ - "event": "scholar_source.search_fetch_started", - "query": query, - "requested_url": requested_url, - "start": start, - }, + structured_log( + logger, "debug", "scholar_source.search_fetch_started", + query=query, + requested_url=requested_url, + start=start, ) return await self._fetch_with_global_throttle(requested_url) async def fetch_publication_html(self, publication_url: str) -> FetchResult: - logger.debug( - "scholar_source.publication_fetch_started", - extra={ - "event": "scholar_source.publication_fetch_started", - "requested_url": publication_url, - }, + structured_log( + logger, "debug", "scholar_source.publication_fetch_started", + requested_url=publication_url, ) return await self._fetch_with_global_throttle(publication_url) @@ -169,9 +161,9 @@ class LiveScholarSource: @staticmethod def _network_error_result(requested_url: str, exc: URLError) -> FetchResult: - logger.warning( - "scholar_source.fetch_network_error", - extra={"event": "scholar_source.fetch_network_error", "requested_url": requested_url}, + structured_log( + logger, "warning", "scholar_source.fetch_network_error", + requested_url=requested_url, ) return FetchResult( requested_url=requested_url, @@ -183,13 +175,10 @@ class LiveScholarSource: @staticmethod def _http_error_result(requested_url: str, exc: HTTPError) -> FetchResult: - logger.warning( - "scholar_source.fetch_http_error", - extra={ - "event": "scholar_source.fetch_http_error", - "requested_url": requested_url, - "status_code": exc.code, - }, + structured_log( + logger, "warning", "scholar_source.fetch_http_error", + requested_url=requested_url, + status_code=exc.code, ) return FetchResult( requested_url=requested_url, @@ -203,13 +192,10 @@ class LiveScholarSource: def _success_result(requested_url: str, response) -> FetchResult: body = response.read().decode("utf-8", errors="replace") status_code = getattr(response, "status", 200) - logger.debug( - "scholar_source.fetch_succeeded", - extra={ - "event": "scholar_source.fetch_succeeded", - "requested_url": requested_url, - "status_code": status_code, - }, + structured_log( + logger, "debug", "scholar_source.fetch_succeeded", + requested_url=requested_url, + status_code=status_code, ) return FetchResult( requested_url=requested_url, diff --git a/app/services/domains/scholars/application.py b/app/services/domains/scholars/application.py index 8da041f..18ffdae 100644 --- a/app/services/domains/scholars/application.py +++ b/app/services/domains/scholars/application.py @@ -41,6 +41,7 @@ from app.services.domains.scholars.constants import ( SEARCH_COOLDOWN_REASON, SEARCH_DISABLED_REASON, ) +from app.logging_utils import structured_log from app.services.domains.scholars.exceptions import ScholarServiceError from app.services.domains.scholars.search_hints import ( _merge_warnings, @@ -430,9 +431,9 @@ def _normalize_author_search_inputs(query: str, limit: int) -> tuple[str, int, s def _disabled_search_result(*, normalized_query: str, bounded_limit: int) -> ParsedAuthorSearchPage: - logger.warning( - "scholar_search.disabled_by_configuration", - extra={"event": "scholar_search.disabled_by_configuration", "query": normalized_query}, + structured_log( + logger, "warning", "scholar_search.disabled_by_configuration", + query=normalized_query, ) return _policy_blocked_author_search_result( reason=SEARCH_DISABLED_REASON, @@ -456,9 +457,9 @@ def _normalize_runtime_cooldown_state( updated = True if now_utc < cooldown_until: return updated - logger.info( - "scholar_search.cooldown_expired", - extra={"event": "scholar_search.cooldown_expired", "cooldown_until_utc": cooldown_until.isoformat()}, + structured_log( + logger, "info", "scholar_search.cooldown_expired", + cooldown_until_utc=cooldown_until.isoformat(), ) runtime_state.cooldown_until = None runtime_state.cooldown_rejection_count = 0 @@ -492,15 +493,12 @@ def _emit_cooldown_threshold_alert( return True if bool(runtime_state.cooldown_alert_emitted): return True - logger.error( - "scholar_search.cooldown_rejection_threshold_exceeded", - extra={ - "event": "scholar_search.cooldown_rejection_threshold_exceeded", - "query": normalized_query, - "cooldown_rejection_count": int(runtime_state.cooldown_rejection_count), - "threshold": threshold, - "cooldown_until_utc": runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, - }, + structured_log( + logger, "error", "scholar_search.cooldown_rejection_threshold_exceeded", + query=normalized_query, + cooldown_rejection_count=int(runtime_state.cooldown_rejection_count), + threshold=threshold, + cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, ) runtime_state.cooldown_alert_emitted = True return True @@ -519,14 +517,11 @@ def _cooldown_block_result( normalized_query=normalized_query, cooldown_rejection_alert_threshold=cooldown_rejection_alert_threshold, ) - logger.warning( - "scholar_search.cooldown_active", - extra={ - "event": "scholar_search.cooldown_active", - "query": normalized_query, - "cooldown_remaining_seconds": cooldown_remaining_seconds, - "cooldown_until_utc": runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, - }, + structured_log( + logger, "warning", "scholar_search.cooldown_active", + query=normalized_query, + cooldown_remaining_seconds=cooldown_remaining_seconds, + cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, ) return _policy_blocked_author_search_result( reason=SEARCH_COOLDOWN_REASON, @@ -553,14 +548,11 @@ async def _cache_hit_result( ) if cached is None: return None - logger.info( - "scholar_search.cache_hit", - extra={ - "event": "scholar_search.cache_hit", - "query": normalized_query, - "state": cached.state.value, - "state_reason": cached.state_reason, - }, + structured_log( + logger, "info", "scholar_search.cache_hit", + query=normalized_query, + state=cached.state.value, + state_reason=cached.state_reason, ) state_reason_override = SEARCH_CACHED_BLOCK_REASON if _is_author_search_block_state(cached) else None return _trim_author_search_result( @@ -610,9 +602,10 @@ async def _wait_for_author_search_throttle( ) if sleep_seconds <= 0.0: return updated - logger.info( - "scholar_search.throttle_wait", - extra={"event": "scholar_search.throttle_wait", "query": normalized_query, "sleep_seconds": round(sleep_seconds, 3)}, + structured_log( + logger, "info", "scholar_search.throttle_wait", + query=normalized_query, + sleep_seconds=round(sleep_seconds, 3), ) await asyncio.sleep(sleep_seconds) return True @@ -665,16 +658,13 @@ def _with_retry_warnings( threshold = max(1, int(retry_alert_threshold)) if retry_scheduled_count < threshold: return merged - logger.warning( - "scholar_search.retry_threshold_exceeded", - extra={ - "event": "scholar_search.retry_threshold_exceeded", - "query": normalized_query, - "retry_scheduled_count": retry_scheduled_count, - "threshold": threshold, - "final_state": merged.state.value, - "final_state_reason": merged.state_reason, - }, + structured_log( + logger, "warning", "scholar_search.retry_threshold_exceeded", + query=normalized_query, + retry_scheduled_count=retry_scheduled_count, + threshold=threshold, + final_state=merged.state.value, + final_state_reason=merged.state_reason, ) return replace( merged, @@ -697,14 +687,11 @@ def _apply_block_circuit_breaker( runtime_state.consecutive_blocked_count = 0 return merged_parsed runtime_state.consecutive_blocked_count = int(runtime_state.consecutive_blocked_count) + 1 - logger.warning( - "scholar_search.block_detected", - extra={ - "event": "scholar_search.block_detected", - "query": normalized_query, - "state_reason": merged_parsed.state_reason, - "consecutive_blocked_count": int(runtime_state.consecutive_blocked_count), - }, + structured_log( + logger, "warning", "scholar_search.block_detected", + query=normalized_query, + state_reason=merged_parsed.state_reason, + consecutive_blocked_count=int(runtime_state.consecutive_blocked_count), ) if int(runtime_state.consecutive_blocked_count) < max(1, int(cooldown_block_threshold)): return merged_parsed @@ -712,13 +699,10 @@ def _apply_block_circuit_breaker( runtime_state.consecutive_blocked_count = 0 runtime_state.cooldown_rejection_count = 0 runtime_state.cooldown_alert_emitted = False - logger.error( - "scholar_search.cooldown_activated", - extra={ - "event": "scholar_search.cooldown_activated", - "query": normalized_query, - "cooldown_until_utc": runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, - }, + structured_log( + logger, "error", "scholar_search.cooldown_activated", + query=normalized_query, + cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, ) return replace( merged_parsed, diff --git a/app/services/domains/unpaywall/application.py b/app/services/domains/unpaywall/application.py index 13104e9..8b4d528 100644 --- a/app/services/domains/unpaywall/application.py +++ b/app/services/domains/unpaywall/application.py @@ -13,6 +13,7 @@ from app.services.domains.unpaywall.pdf_discovery import ( resolve_pdf_from_landing_page, ) from app.services.domains.unpaywall.rate_limit import wait_for_unpaywall_slot +from app.logging_utils import structured_log from app.settings import settings if TYPE_CHECKING: @@ -157,10 +158,7 @@ async def _resolved_pdf_url_from_payload( for page_url in _crawl_targets(payload=payload, pdf_candidates=pdf_candidates)[:3]: discovered = await resolve_pdf_from_landing_page(client, page_url=page_url) if discovered: - logger.info( - "unpaywall.pdf_discovered_from_landing", - extra={"event": "unpaywall.pdf_discovered_from_landing", "landing_url": page_url}, - ) + structured_log(logger, "info", "unpaywall.pdf_discovered_from_landing", landing_url=page_url) return discovered return None @@ -191,27 +189,6 @@ def _email_for_request(request_email: str | None) -> str | None: return email or None -def _log_resolution_summary( - *, - publication_count: int, - doi_input_count: int, - search_attempt_count: int, - resolved_pdf_count: int, - email: str, -) -> None: - logger.info( - "unpaywall.resolve_completed", - extra={ - "event": "unpaywall.resolve_completed", - "publication_count": publication_count, - "doi_input_count": doi_input_count, - "search_attempt_count": search_attempt_count, - "resolved_pdf_count": resolved_pdf_count, - "email_domain": email.split("@", 1)[-1] if "@" in email else None, - }, - ) - - async def _resolve_item_payload( *, client, @@ -370,14 +347,7 @@ async def _safe_outcome_for_item( allow_crossref=allow_crossref, ) except Exception as exc: # pragma: no cover - defensive network boundary - logger.warning( - "unpaywall.resolve_item_failed", - extra={ - "event": "unpaywall.resolve_item_failed", - "publication_id": item.publication_id, - "error": str(exc), - }, - ) + structured_log(logger, "warning", "unpaywall.resolve_item_failed", publication_id=item.publication_id, error=str(exc)) return _outcome_with_failure( item=item, failure_reason=FAILURE_RESOLUTION_EXCEPTION, @@ -439,12 +409,13 @@ async def resolve_publication_oa_outcomes( targets=targets, email=email, ) - _log_resolution_summary( + structured_log( + logger, "info", "unpaywall.resolve_completed", publication_count=len(items), doi_input_count=_doi_input_count(items), search_attempt_count=_search_attempt_count(targets=targets), resolved_pdf_count=_resolved_pdf_count(outcomes), - email=email, + email_domain=email.split("@", 1)[-1] if "@" in email else None, ) return outcomes diff --git a/tests/unit/services/domains/arxiv/test_client.py b/tests/unit/services/domains/arxiv/test_client.py index f1c4c51..445d3cd 100644 --- a/tests/unit/services/domains/arxiv/test_client.py +++ b/tests/unit/services/domains/arxiv/test_client.py @@ -147,9 +147,7 @@ async def test_client_logs_cache_hit_and_miss( return httpx.Response(200, text=_CLIENT_FEED_XML, request=request) def _capture_log(_msg: str, *args, **kwargs) -> None: - extra = kwargs.get("extra") - if isinstance(extra, dict): - logged.append(extra) + logged.append({"event": _msg, **(kwargs.get("extra") or {})}) monkeypatch.setattr("app.services.domains.arxiv.client.logger.info", _capture_log) client = ArxivClient(request_fn=_request_fn, cache_enabled=True) @@ -183,9 +181,7 @@ async def test_request_feed_skips_live_call_when_global_cooldown_is_active( return httpx.Response(200, text=_CLIENT_FEED_XML) def _capture_warning(_msg: str, *args, **kwargs) -> None: - extra = kwargs.get("extra") - if isinstance(extra, dict): - logged.append(extra) + logged.append({"event": _msg, **(kwargs.get("extra") or {})}) monkeypatch.setattr(arxiv_client_module, "get_arxiv_cooldown_status", _cooldown_status) monkeypatch.setattr(arxiv_client_module, "run_with_global_arxiv_limit", _unexpected_limit_call) diff --git a/tests/unit/services/domains/arxiv/test_rate_limit.py b/tests/unit/services/domains/arxiv/test_rate_limit.py index 4e5623b..86ad949 100644 --- a/tests/unit/services/domains/arxiv/test_rate_limit.py +++ b/tests/unit/services/domains/arxiv/test_rate_limit.py @@ -95,9 +95,7 @@ async def test_arxiv_rate_limit_logs_request_scheduled_and_completed( return httpx.Response(200, text="ok") def _capture_log(_msg: str, *args, **kwargs) -> None: - extra = kwargs.get("extra") - if isinstance(extra, dict): - logged.append(extra) + logged.append({"event": _msg, **(kwargs.get("extra") or {})}) monkeypatch.setattr("app.services.domains.arxiv.rate_limit.logger.info", _capture_log) await run_with_global_arxiv_limit(fetch=_fetch, source_path="search") @@ -126,9 +124,7 @@ async def test_arxiv_rate_limit_logs_cooldown_activation( return httpx.Response(429, text="rate limited") def _capture_warning(_msg: str, *args, **kwargs) -> None: - extra = kwargs.get("extra") - if isinstance(extra, dict): - logged_warning.append(extra) + logged_warning.append({"event": _msg, **(kwargs.get("extra") or {})}) monkeypatch.setattr("app.services.domains.arxiv.rate_limit.logger.warning", _capture_warning) with pytest.raises(ArxivRateLimitError):