gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(tasks): address code review feedback for notification dispatch

Browse Source 
- Move `import os` to module level in celery_app.py
- Add detailed task signature comments to _FILE_ID_ARG_INDEX mapping
- Upgrade failure dispatch log level from debug to warning

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-16 16:23:06 +00:00
parent 195c3c3446
commit 803331d61f
1 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/celery_app.py
+9 -6
View File
@@ -1,6 +1,7 @@
# app/celery_app.py # app/celery_app.py
import logging import logging
import os
from celery import Celery from celery import Celery
from celery.signals import task_failure, worker_ready from celery.signals import task_failure, worker_ready
@@ -26,9 +27,13 @@ celery.conf.task_routes = {
} }
# Mapping of document pipeline task names to the positional index of ``file_id`` # Mapping of document pipeline task names to the positional index of ``file_id``
# in their ``args`` tuple. Tasks that always pass ``file_id`` as a keyword # in their ``args`` tuple. These indices correspond to the task signatures:
# argument (e.g. ``process_document``, ``finalize_document_storage``) are not # process_with_ocr(filename, file_id, ...) → index 1
# listed here — their ``file_id`` is found via ``kwargs`` instead. # extract_metadata_with_gpt(filename, text, file_id) → index 2
# embed_metadata_into_pdf(path, text, metadata, file_id) → index 3
# Tasks that always pass ``file_id`` as a keyword argument
# (e.g. ``process_document``, ``finalize_document_storage``) are not listed
# here — their ``file_id`` is found via ``kwargs`` instead.
_FILE_ID_ARG_INDEX: dict[str, int] = { _FILE_ID_ARG_INDEX: dict[str, int] = {
"app.tasks.process_with_ocr.process_with_ocr": 1, "app.tasks.process_with_ocr.process_with_ocr": 1,
"app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt": 2, "app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt": 2,
@@ -71,8 +76,6 @@ def _dispatch_user_failure_notification(sender, exception, args: list | None, kw
filename = record.original_filename or record.local_filename or "unknown" filename = record.original_filename or record.local_filename or "unknown"
# 3. Dispatch per-user notification # 3. Dispatch per-user notification
import os
error_msg = f"{type(exception).__name__}: {exception}" if exception else "Unknown error" error_msg = f"{type(exception).__name__}: {exception}" if exception else "Unknown error"
notify_user_document_failed( notify_user_document_failed(
owner_id=owner_id, owner_id=owner_id,
@@ -114,4 +117,4 @@ def task_failure_handler(
try: try:
_dispatch_user_failure_notification(sender, exception, args, kwargs) _dispatch_user_failure_notification(sender, exception, args, kwargs)
except Exception: except Exception:
logger.debug("Could not dispatch per-user failure notification", exc_info=True) logger.warning("Could not dispatch per-user failure notification", exc_info=True)