v6ole
2996920277
- 添加 LogHiveHandler，启动时自动挂载到 root logger - 所有 logging 日志自动异步发送到 LogHive，不影响主业务 - vendored loghive-client 包，Docker 构建时自动安装 - API Key 缺失时自动跳过，不影响本地开发
73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
"""Python logging.Handler integration — use LogHive with the stdlib logging module.
|
|
|
|
This allows you to replace or augment your existing logging setup with
|
|
zero code changes (just add a handler to your logger).
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from loghive_client.client import LogHiveLogger
|
|
|
|
|
|
class LogHiveHandler(logging.Handler):
|
|
"""A logging.Handler that sends records to LogHive.
|
|
|
|
Use it with Python's standard logging module:
|
|
import logging
|
|
from loghive_client import LogHiveHandler
|
|
|
|
handler = LogHiveHandler("my-project", "api-key", "http://localhost:8000")
|
|
logging.getLogger().addHandler(handler)
|
|
|
|
All existing logger calls (logger.info, logger.error, etc.) will
|
|
automatically forward to LogHive.
|
|
"""
|
|
|
|
LEVEL_MAP = {
|
|
logging.DEBUG: "debug",
|
|
logging.INFO: "info",
|
|
logging.WARNING: "warning",
|
|
logging.ERROR: "error",
|
|
logging.CRITICAL: "critical",
|
|
}
|
|
|
|
def __init__(
|
|
self,
|
|
project: str,
|
|
api_key: str,
|
|
endpoint: str = "http://localhost:8000",
|
|
level: int = logging.INFO,
|
|
):
|
|
super().__init__(level=level)
|
|
self._client = LogHiveLogger(
|
|
project=project,
|
|
api_key=api_key,
|
|
endpoint=endpoint,
|
|
)
|
|
|
|
def emit(self, record: logging.LogRecord):
|
|
"""Send a log record to LogHive."""
|
|
try:
|
|
level = self.LEVEL_MAP.get(record.levelno, "info")
|
|
extra = {
|
|
"logger": record.name,
|
|
"module": record.module,
|
|
"function": record.funcName,
|
|
"line_no": record.lineno,
|
|
}
|
|
if record.exc_info and record.exc_info[0]:
|
|
import traceback
|
|
extra["exception"] = "".join(
|
|
traceback.format_exception(*record.exc_info)
|
|
)
|
|
|
|
self._client._enqueue(level, record.getMessage(), **extra)
|
|
except Exception:
|
|
self.handleError(record)
|
|
|
|
def close(self):
|
|
"""Flush and close."""
|
|
self._client.stop(flush=True)
|
|
super().close()
|