Coverage for astrocyte/ingest/logutil.py: 100%

1"""Structured logging helpers for ingest (poll, stream, supervisor). 

2 

3When ``ASTROCYTE_LOG_FORMAT`` is ``json`` / ``1`` / ``true`` / ``yes`` (same convention as 

4``astrocyte-gateway`` :mod:`astrocyte_gateway.observability`), emit one JSON object per line on 

5the given logger at INFO. Otherwise emit a short human-readable line. 

6 

7Environment is read at **call time** so tests and workers can toggle without import order issues. 

8""" 

9 

10from __future__ import annotations 

11 

12import json 

13import logging 

14import os 

15from typing import Any 

16 

17 

18def _json_logs_enabled() -> bool: 

19 return os.environ.get("ASTROCYTE_LOG_FORMAT", "").strip().lower() in ("json", "1", "true", "yes") 

20 

21 

22def log_ingest_event(logger: logging.Logger, event: str, **fields: Any) -> None: 

23 """Log a single ingest observability event (supervisor lifecycle, rate limits, transport errors).""" 

24 if _json_logs_enabled(): 

25 payload: dict[str, Any] = {"event": event, **fields} 

26 logger.info(json.dumps(payload, ensure_ascii=False, default=str)) 

27 return 

28 parts = " ".join(f"{k}={v!r}" for k, v in fields.items()) 

29 logger.info("%s %s", event, parts)