Initialize async client with HTTP or Unix-socket transport. Args: endpoint: HTTP/HTTPS URL or Unix socket path use_sync_http: If True, use sync HTTP client internally timeout: Timeout in seconds
(
self,
endpoint: str | None = None,
*,
use_sync_http: bool = False,
timeout: float = 3,
)
| 283 | PATH_PREFIX = "/" |
| 284 | |
| 285 | def __init__( |
| 286 | self, |
| 287 | endpoint: str | None = None, |
| 288 | *, |
| 289 | use_sync_http: bool = False, |
| 290 | timeout: float = 3, |
| 291 | ): |
| 292 | """Initialize async client with HTTP or Unix-socket transport. |
| 293 | |
| 294 | Args: |
| 295 | endpoint: HTTP/HTTPS URL or Unix socket path |
| 296 | use_sync_http: If True, use sync HTTP client internally |
| 297 | timeout: Timeout in seconds |
| 298 | |
| 299 | """ |
| 300 | endpoint = get_endpoint(endpoint) |
| 301 | self.use_sync_http = use_sync_http |
| 302 | self._client: Optional[httpx.AsyncClient] = None |
| 303 | self._sync_client: Optional[httpx.Client] = None |
| 304 | self._client_ref_count = 0 |
| 305 | self._timeout = timeout |
| 306 | |
| 307 | if endpoint.startswith("http://") or endpoint.startswith("https://"): |
| 308 | self.async_transport = httpx.AsyncHTTPTransport() |
| 309 | self.sync_transport = httpx.HTTPTransport() |
| 310 | self.base_url = endpoint |
| 311 | else: |
| 312 | # Check if Unix socket file exists |
| 313 | if endpoint.startswith("/") and not os.path.exists(endpoint): |
| 314 | raise FileNotFoundError(f"Unix socket file {endpoint} does not exist") |
| 315 | self.async_transport = httpx.AsyncHTTPTransport(uds=endpoint) |
| 316 | self.sync_transport = httpx.HTTPTransport(uds=endpoint) |
| 317 | self.base_url = "http://localhost" |
| 318 | |
| 319 | def _get_client(self) -> httpx.AsyncClient: |
| 320 | if self._client is None: |
nothing calls this directly
no test coverage detected