objectif
Produire des logs JSON parsables pour ingestion centralisée.
code minimal
import json, logging, io
class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
payload = {
"level": record.levelname,
"name": record.name,
"message": record.getMessage(),
}
return json.dumps(payload, ensure_ascii=False)
stream = io.StringIO()
handler = logging.StreamHandler(stream)
handler.setFormatter(JsonFormatter())
log = logging.getLogger("svc")
log.setLevel(logging.INFO)
log.handlers = [handler]
log.info("hello")
line = stream.getvalue().strip()
print(json.loads(line)["message"]) # attendu: hello
utilisation
import json
s = '{"level":"INFO","name":"svc","message":"ok"}'
print(json.loads(s)["level"] == "INFO")
variante(s) utile(s)
import json, logging, io
class JsonFormatter(logging.Formatter):
def format(self, record):
return json.dumps({"ts": record.created, "msg": record.getMessage()}, ensure_ascii=False)
buf = io.StringIO()
h = logging.StreamHandler(buf); h.setFormatter(JsonFormatter())
lg = logging.getLogger("x"); lg.setLevel(logging.INFO); lg.handlers=[h]
lg.info("x")
print("ts" in buf.getvalue())
notes
- Ajoutez timestamp, thread, correlation-id selon vos besoins.
- Évitez print() pour du logging en production.