Asynchronously checks the health of the configured API. This method is used in conjunction with require_api_key() to verify that the API is not just configured, but also responsive. It makes a test request to a known endpoint to validate the API's health. The method uses the `ping_
(self, url=None)
| 366 | return (self.api_retries * self._api_failure_abort_threshold) + 1 |
| 367 | |
| 368 | async def ping(self, url=None): |
| 369 | """Asynchronously checks the health of the configured API. |
| 370 | |
| 371 | This method is used in conjunction with require_api_key() to verify that the API is not just configured, but also responsive. It makes a test request to a known endpoint to validate the API's health. |
| 372 | |
| 373 | The method uses the `ping_url` attribute if defined, or falls back to a provided URL. If neither is available, no request is made. |
| 374 | |
| 375 | Args: |
| 376 | url (str, optional): A specific URL to use for the ping request. If not provided, the method will use the `ping_url` attribute. |
| 377 | |
| 378 | Returns: |
| 379 | None |
| 380 | |
| 381 | Raises: |
| 382 | ValueError: If the API response is not successful (status code != 200). |
| 383 | |
| 384 | Example Usage: |
| 385 | To use this method, simply define the `ping_url` attribute in your module: |
| 386 | |
| 387 | class MyModule(BaseModule): |
| 388 | ping_url = "https://api.example.com/ping" |
| 389 | |
| 390 | Alternatively, you can override this method for more complex health checks: |
| 391 | |
| 392 | async def ping(self): |
| 393 | r = await self.api_request(f"{self.base_url}/complex-health-check") |
| 394 | if r.status_code != 200 or r.json().get('status') != 'healthy': |
| 395 | raise ValueError(f"API unhealthy: {r.text}") |
| 396 | """ |
| 397 | if url is None: |
| 398 | url = getattr(self, "ping_url", "") |
| 399 | retry_on_http_429 = getattr(self, "_ping_retry_on_http_429", False) |
| 400 | if url: |
| 401 | r = await self.api_request(url, retry_on_http_429=retry_on_http_429) |
| 402 | if getattr(r, "status_code", 0) != 200: |
| 403 | response_text = getattr(r, "text", "no response from server") |
| 404 | raise ValueError(response_text) |
| 405 | |
| 406 | @property |
| 407 | def batch_size(self): |
no test coverage detected