Start the async authentication task.
(self, api_key: str)
| 116 | return None |
| 117 | |
| 118 | def _start_auth_task(self, api_key: str): |
| 119 | """Start the async authentication task.""" |
| 120 | if self._auth_task and not self._auth_task.done(): |
| 121 | return # Task already running |
| 122 | |
| 123 | try: |
| 124 | loop = asyncio.get_event_loop() |
| 125 | if loop.is_running(): |
| 126 | # Use existing event loop |
| 127 | self._auth_task = loop.create_task(self._fetch_auth_async(api_key)) |
| 128 | else: |
| 129 | # Create new event loop in background thread |
| 130 | def run_async_auth(): |
| 131 | asyncio.run(self._fetch_auth_async(api_key)) |
| 132 | |
| 133 | import threading |
| 134 | |
| 135 | auth_thread = threading.Thread(target=run_async_auth, daemon=True) |
| 136 | auth_thread.start() |
| 137 | except RuntimeError: |
| 138 | # Create new event loop in background thread |
| 139 | def run_async_auth(): |
| 140 | asyncio.run(self._fetch_auth_async(api_key)) |
| 141 | |
| 142 | import threading |
| 143 | |
| 144 | auth_thread = threading.Thread(target=run_async_auth, daemon=True) |
| 145 | auth_thread.start() |
| 146 | |
| 147 | def init(self, **kwargs: Any) -> None: # Return type updated to None |
| 148 | # Recreate the Config object to parse environment variables at the time of initialization |
no test coverage detected