-
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlogger_utils.py
More file actions
67 lines (55 loc) · 2.18 KB
/
Copy pathlogger_utils.py
File metadata and controls
67 lines (55 loc) · 2.18 KB
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
# docs: docs/guides/developing-finecode.md
import inspect
import logging
import sys
from pathlib import Path
from loguru import logger
from finecode_extension_runner import logs
def init_logger(
log_name: str,
log_level: str = "INFO",
stdout: bool = False,
log_groups: dict[str, str] | None = None,
workspace_path: Path | None = None,
otlp_endpoint: str | None = None,
) -> Path:
venv_dir_path = Path(sys.executable).parent.parent
logs_dir_path = venv_dir_path / "logs"
logger.remove()
log_file_path = logs.save_logs_to_file(
file_path=logs_dir_path / log_name / f"{log_name}.log",
log_level=log_level,
stdout=stdout,
)
if log_groups:
for group, level_str in log_groups.items():
try:
logs.set_log_level_for_group(group, logs.LogLevel[level_str.upper()])
except KeyError:
pass
# pygls uses standard python logger, intercept it and pass logs to loguru
class InterceptHandler(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
# Get corresponding Loguru level if it exists.
level: str | int
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
# Find caller from where originated the logged message.
frame, depth = inspect.currentframe(), 0
while frame and (
depth == 0 or frame.f_code.co_filename == logging.__file__
):
frame = frame.f_back
depth += 1
logger.opt(depth=depth, exception=record.exc_info).log(
level, record.getMessage()
)
logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
from finecode import telemetry
service_name = f"finecode-{log_name.replace('_', '-')}"
telemetry.init_otel_logging(service_name, workspace_path, endpoint=otlp_endpoint)
telemetry.init_tracer_provider(service_name, workspace_path, endpoint=otlp_endpoint)
telemetry.init_meter_provider(service_name, workspace_path, endpoint=otlp_endpoint)
return log_file_path