Singleton client for AgentOps service
| 36 | |
| 37 | |
| 38 | class Client: |
| 39 | """Singleton client for AgentOps service""" |
| 40 | |
| 41 | config: Config |
| 42 | _initialized: bool |
| 43 | _init_trace_context: Optional[TraceContext] = None # Stores the context of the auto-started trace |
| 44 | _legacy_session_for_init_trace: Optional[Session] = ( |
| 45 | None # Stores the legacy Session wrapper for the auto-started trace |
| 46 | ) |
| 47 | |
| 48 | __instance = None # Class variable for singleton pattern |
| 49 | |
| 50 | api: ApiClient |
| 51 | _auth_token: Optional[str] = None |
| 52 | _project_id: Optional[str] = None |
| 53 | _auth_lock = threading.Lock() |
| 54 | _auth_task: Optional[asyncio.Task] = None |
| 55 | |
| 56 | def __new__(cls, *args: Any, **kwargs: Any) -> "Client": |
| 57 | if cls.__instance is None: |
| 58 | cls.__instance = super(Client, cls).__new__(cls) |
| 59 | # Initialize instance variables that should only be set once per instance |
| 60 | cls.__instance._init_trace_context = None |
| 61 | cls.__instance._legacy_session_for_init_trace = None |
| 62 | cls.__instance._auth_token = None |
| 63 | cls.__instance._project_id = None |
| 64 | cls.__instance._auth_lock = threading.Lock() |
| 65 | cls.__instance._auth_task = None |
| 66 | return cls.__instance |
| 67 | |
| 68 | def __init__(self): |
| 69 | # Initialization of attributes like config, _initialized should happen here if they are instance-specific |
| 70 | # and not shared via __new__ for a true singleton that can be re-configured. |
| 71 | # However, the current pattern re-initializes config in init(). |
| 72 | if ( |
| 73 | not hasattr(self, "_initialized") or not self._initialized |
| 74 | ): # Ensure init logic runs only once per actual initialization intent |
| 75 | self.config = Config() # Initialize config here for the instance |
| 76 | self._initialized = False |
| 77 | # self._init_trace_context = None # Already done in __new__ |
| 78 | # self._legacy_session_for_init_trace = None # Already done in __new__ |
| 79 | |
| 80 | def get_current_jwt(self) -> Optional[str]: |
| 81 | """Get the current JWT token.""" |
| 82 | with self._auth_lock: |
| 83 | return self._auth_token |
| 84 | |
| 85 | def _set_auth_data(self, token: str, project_id: str): |
| 86 | """Set authentication data thread-safely.""" |
| 87 | with self._auth_lock: |
| 88 | self._auth_token = token |
| 89 | self._project_id = project_id |
| 90 | |
| 91 | # Update the HTTP client's project ID |
| 92 | from agentops.client.http.http_client import HttpClient |
| 93 | |
| 94 | HttpClient.set_project_id(project_id) |
| 95 |
no outgoing calls
searching dependent graphs…