langfuse.utils.langfuse_singleton

 1import threading
 2from typing import Optional
 3
 4import httpx
 5
 6from langfuse import Langfuse
 7from langfuse.types import MaskFunction
 8
 9
10class LangfuseSingleton:
11    _instance = None
12    _lock = threading.Lock()
13    _langfuse: Optional[Langfuse] = None
14
15    def __new__(cls):
16        if not cls._instance:
17            with cls._lock:
18                if not cls._instance:
19                    cls._instance = super(LangfuseSingleton, cls).__new__(cls)
20        return cls._instance
21
22    def get(
23        self,
24        *,
25        public_key: Optional[str] = None,
26        secret_key: Optional[str] = None,
27        host: Optional[str] = None,
28        release: Optional[str] = None,
29        debug: Optional[bool] = None,
30        threads: Optional[int] = None,
31        flush_at: Optional[int] = None,
32        flush_interval: Optional[int] = None,
33        max_retries: Optional[int] = None,
34        timeout: Optional[int] = None,
35        httpx_client: Optional[httpx.Client] = None,
36        sdk_integration: Optional[str] = None,
37        enabled: Optional[bool] = None,
38        sample_rate: Optional[float] = None,
39        mask: Optional[MaskFunction] = None,
40        environment: Optional[str] = None,
41    ) -> Langfuse:
42        if self._langfuse:
43            return self._langfuse
44
45        with self._lock:
46            if self._langfuse:
47                return self._langfuse
48
49            langfuse_init_args = {
50                "public_key": public_key,
51                "secret_key": secret_key,
52                "host": host,
53                "release": release,
54                "debug": debug,
55                "threads": threads,
56                "flush_at": flush_at,
57                "flush_interval": flush_interval,
58                "max_retries": max_retries,
59                "timeout": timeout,
60                "httpx_client": httpx_client,
61                "sdk_integration": sdk_integration,
62                "enabled": enabled,
63                "sample_rate": sample_rate,
64                "mask": mask,
65                "environment": environment,
66            }
67
68            self._langfuse = Langfuse(
69                **{k: v for k, v in langfuse_init_args.items() if v is not None}
70            )
71
72            return self._langfuse
73
74    def reset(self) -> None:
75        with self._lock:
76            if self._langfuse:
77                self._langfuse.shutdown()
78
79            self._langfuse = None