Код: Выделить всё
{
"msg": "some event",
"level": "warning",
"timestamp": "2026-03-09 19:01:20",
"user_id": 123,
"frequency": "some key"
// some other key-value pairs
}
Код: Выделить всё
def configure_events() -> None:
processors = [
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
structlog.stdlib.render_to_log_kwargs,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.JSONRenderer()
]
structlog.configure(
processors=processors,
wrapper_class=structlog.make_filtering_bound_logger("INFO"),
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
Код: Выделить всё
@pytest.fixture
def capture_logs():
emitted = []
def capture(*args):
event_dict = args[-1]
emitted.append(event_dict)
return event_dict
with patch("structlog.processors.JSONRenderer.__call__", capture):
yield emitted
Код: Выделить всё
def test_event_emits_structured_wide_event(capture_logs):
configure_events()
logger = structlog.get_logger()
logger.info("some event", user_id=123)
assert len(capture_logs) == 1
log = capture_logs[0]
assert log["msg"] == "some event"
assert log["extra"]["level"] == "info"
assert log["extra"]["user_id"] == 123
Но я не хочу, чтобы это было так, я хочу, чтобы все ключи были в виде простого JSON. Я попробовал много разных подходов и в конце концов придумал добавить собственный процессор, который сглаживает значение дополнительного ключа, поэтому после реализации моя конфигурация structlog выглядит следующим образом:
Код: Выделить всё
def configure_events() -> None:
def _flatten_extra(logger, method_name, event_dict):
if "extra" in event_dict:
extra = event_dict.pop("extra")
if not isinstance(extra, dict):
raise ValueError("Value of 'extra' is not of dict type.")
event_dict.update(extra)
return event_dict
processors = [
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
structlog.stdlib.render_to_log_kwargs,
structlog.stdlib.PositionalArgumentsFormatter(),
_flatten_extra,
structlog.processors.JSONRenderer()
]
structlog.configure(
processors=processors,
wrapper_class=structlog.make_filtering_bound_logger("INFO"),
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
Но это не работает с ПРЕДУПРЕЖДЕНИЕМ.
У меня есть такой тест:
Код: Выделить всё
def test_event_can_use_warning_level(capture_logs):
configure_events()
logger = structlog.get_logger()
logger.warning("check_failed", user_id=222, reason="timeout")
assert capture_logs[0]["level"] == "warning"
Код: Выделить всё
def test_event_can_use_warning_level(capture_logs):
configure_events()
logger = structlog.get_logger()
extra = {"user_id": 222, "reason": "timeout"}
logger.warning("check_failed", extra=extra)
assert capture_logs[0]["level"] == "warning"
Я здесь совершенно потерян, и у меня нет идей. Что я упускаю или неправильно настраиваю?
Подробнее здесь: https://stackoverflow.com/questions/799 ... de-logging
Мобильная версия