Repository navigation
Expand file tree
/
Copy pathtelemetry.py
More file actions
64 lines (48 loc) · 1.68 KB
/
Copy pathtelemetry.py
File metadata and controls
64 lines (48 loc) · 1.68 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
"""Logfire telemetry bootstrap.
`configure_telemetry()` wires up the Logfire SDK and returns the loguru
handler. Call sites use `logfire.span(...)` and `logfire.metric_counter(...)`
directly — there are no wrappers here.
"""
from __future__ import annotations
from typing import Any
import logfire
REPOSITORY_URL = "https://github.com/basicmachines-co/basic-memory"
ROOT_PATH = "src/basic_memory"
_LOGFIRE_HANDLER: dict[str, Any] | None = None
def configure_telemetry(
service_name: str,
*,
environment: str,
service_version: str | None = None,
enable_logfire: bool = False,
send_to_logfire: bool = False,
log_level: str = "INFO",
) -> bool:
"""Configure Logfire for the current process. Returns True when enabled."""
global _LOGFIRE_HANDLER
_LOGFIRE_HANDLER = None
if not enable_logfire:
return False
kwargs: dict[str, Any] = {
"service_name": service_name,
"environment": environment,
"code_source": logfire.CodeSource(
repository=REPOSITORY_URL,
revision=service_version or "",
root_path=ROOT_PATH,
),
"min_level": log_level.lower(),
"send_to_logfire": send_to_logfire,
}
try:
logfire.configure(**kwargs)
except TypeError:
# Older logfire releases don't accept send_to_logfire as a keyword.
kwargs.pop("send_to_logfire", None)
logfire.configure(**kwargs)
_LOGFIRE_HANDLER = logfire.loguru_handler()
return True
def get_logfire_handler() -> dict[str, Any] | None:
"""Return the active Logfire loguru handler, if any."""
return _LOGFIRE_HANDLER
__all__ = ["configure_telemetry", "get_logfire_handler"]