Coverage for paperless/logging.py: 100%

1from __future__ import annotations 

2 

3import logging 

4from contextvars import ContextVar 

5 

6consume_task_id: ContextVar[str] = ContextVar("consume_task_id", default="") 

7 

8 

9class ConsumeTaskFormatter(logging.Formatter): 

10 """ 

11 Logging formatter that prepends a short task correlation ID to messages 

12 emitted during document consumption. 

13 

14 The ID is the first 8 characters of the Celery task UUID, set via the 

15 ``consume_task_id`` ContextVar at the entry of ``consume_file``. When 

16 the ContextVar is empty (any log outside a consume task) no prefix is 

17 added and the output is identical to the standard verbose format. 

18 """ 

19 

20 def __init__(self) -> None: 

21 super().__init__( 

22 fmt="[{asctime}] [{levelname}] [{name}] {task_prefix}{message}", 

23 style="{", 

24 validate=False, # {task_prefix} is not a standard LogRecord attribute, so Python's 

25 # init-time format-string validation would raise ValueError without 

26 # this. Runtime safety comes from format() always setting 

27 # record.task_prefix before calling super().format(). 

28 ) 

29 

30 def format(self, record: logging.LogRecord) -> str: 

31 task_id = consume_task_id.get() 

32 record.task_prefix = f"[{task_id}] " if task_id else "" 

33 return super().format(record)