File size: 3,795 Bytes
aac350d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """
Structured logging with execution context.
Every execution (a job, a provider invocation, a retry attempt) is
given an `ExecutionContext` that binds the following fields into every
log line produced inside that context:
execution_id — UUID for the top-level job
provider_id — name of the provider being invoked (or "-")
retry_count — how many retries have occurred for this attempt
status — "started" | "success" | "failed" | "retried"
Implementation: loguru's `contextualize()` binds dict keys into the
`extra` dict of every record created within the context. We layer a
small wrapper on top so callers don't have to remember the keys.
"""
from __future__ import annotations
import sys
import uuid
from contextlib import contextmanager
from contextvars import ContextVar
from typing import Any, Iterator, Optional
from loguru import logger
from config.settings import Settings
# Context vars for in-process propagation (e.g. into async tasks)
_execution_id: ContextVar[str] = ContextVar("execution_id", default="-")
_provider_id: ContextVar[str] = ContextVar("provider_id", default="-")
_retry_count: ContextVar[int] = ContextVar("retry_count", default=0)
_status: ContextVar[str] = ContextVar("status", default="started")
def setup_logging(settings: Settings) -> None:
"""Configure loguru sinks. Call once at app startup."""
logger.remove()
if settings.log_json:
fmt = (
'{{"timestamp":"{time:YYYY-MM-DDTHH:mm:ss.SSSZ}",'
'"level":"{level}",'
'"execution_id":"{extra[execution_id]}",'
'"provider_id":"{extra[provider_id]}",'
'"retry_count":{extra[retry_count]},'
'"status":"{extra[status]}",'
'"message":"{message}"}}'
)
else:
fmt = (
"{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | "
"eid={extra[execution_id]} | pid={extra[provider_id]} | "
"retry={extra[retry_count]} | status={extra[status]} | "
"{message}"
)
logger.configure(
extra={
"execution_id": "-",
"provider_id": "-",
"retry_count": 0,
"status": "started",
}
)
logger.add(sys.stderr, format=fmt, level=settings.log_level, backtrace=True, diagnose=False)
def new_execution_id() -> str:
"""Generate a fresh execution id."""
return uuid.uuid4().hex[:12]
@contextmanager
def execution_context(
execution_id: Optional[str] = None,
provider_id: str = "-",
retry_count: int = 0,
status: str = "started",
) -> Iterator[dict]:
"""Bind execution context to every log line produced inside the block.
Usage:
with execution_context(provider_id="haar") as ctx:
ctx["status"] = "running"
logger.info("detecting faces")
ctx["status"] = "success"
"""
eid = execution_id or new_execution_id()
ctx = {
"execution_id": eid,
"provider_id": provider_id,
"retry_count": retry_count,
"status": status,
}
eid_token = _execution_id.set(eid)
pid_token = _provider_id.set(provider_id)
rc_token = _retry_count.set(retry_count)
st_token = _status.set(status)
with logger.contextualize(**ctx):
try:
yield ctx
finally:
_execution_id.reset(eid_token)
_provider_id.reset(pid_token)
_retry_count.reset(rc_token)
_status.reset(st_token)
def current_execution_id() -> str:
return _execution_id.get()
def current_provider_id() -> str:
return _provider_id.get()
__all__ = [
"setup_logging",
"new_execution_id",
"execution_context",
"current_execution_id",
"current_provider_id",
"logger",
]
|