langfuse.utils.langfuse_singleton

 1import httpx
 2import threading
 3from typing import Optional
 4
 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    ) -> Langfuse:
41        if self._langfuse:
42            return self._langfuse
43
44        with self._lock:
45            if self._langfuse:
46                return self._langfuse
47
48            langfuse_init_args = {
49                "public_key": public_key,
50                "secret_key": secret_key,
51                "host": host,
52                "release": release,
53                "debug": debug,
54                "threads": threads,
55                "flush_at": flush_at,
56                "flush_interval": flush_interval,
57                "max_retries": max_retries,
58                "timeout": timeout,
59                "httpx_client": httpx_client,
60                "sdk_integration": sdk_integration,
61                "enabled": enabled,
62                "sample_rate": sample_rate,
63                "mask": mask,
64            }
65
66            self._langfuse = Langfuse(
67                **{k: v for k, v in langfuse_init_args.items() if v is not None}
68            )
69
70            return self._langfuse
71
72    def reset(self) -> None:
73        with self._lock:
74            if self._langfuse:
75                self._langfuse.flush()
76
77            self._langfuse = None