Make an HTTP request to the server. Args: method: HTTP method (GET, POST, etc.) path: URL path headers: HTTP headers body: Request body (will be JSON serialized if a dict) stream: If True, return a file-like object for reading the
(
self,
method: str,
path: str,
headers: Dict[str, str] = None,
body: Any = None,
stream: bool = False,
)
| 387 | self.is_https = self.parsed_url.scheme == "https" |
| 388 | |
| 389 | def request( |
| 390 | self, |
| 391 | method: str, |
| 392 | path: str, |
| 393 | headers: Dict[str, str] = None, |
| 394 | body: Any = None, |
| 395 | stream: bool = False, |
| 396 | ) -> Tuple[int, Union[Dict, str, BinaryIO]]: |
| 397 | """Make an HTTP request to the server. |
| 398 | |
| 399 | Args: |
| 400 | method: HTTP method (GET, POST, etc.) |
| 401 | path: URL path |
| 402 | headers: HTTP headers |
| 403 | body: Request body (will be JSON serialized if a dict) |
| 404 | stream: If True, return a file-like object for reading the response |
| 405 | |
| 406 | Returns: |
| 407 | Tuple of (status_code, response_data) |
| 408 | |
| 409 | """ |
| 410 | if headers is None: |
| 411 | headers = {} |
| 412 | |
| 413 | # Add Basic Authentication header if credentials are provided |
| 414 | if self.auth_user and self.auth_password: |
| 415 | credentials = f"{self.auth_user}:{self.auth_password}" |
| 416 | encoded_credentials = base64.b64encode(credentials.encode("utf-8")).decode( |
| 417 | "ascii" |
| 418 | ) |
| 419 | headers["Authorization"] = f"Basic {encoded_credentials}" |
| 420 | |
| 421 | # Prepare the body |
| 422 | if isinstance(body, dict): |
| 423 | body = json.dumps(body).encode("utf-8") |
| 424 | if "Content-Type" not in headers: |
| 425 | headers["Content-Type"] = "application/json" |
| 426 | |
| 427 | # Create the appropriate connection |
| 428 | if self.use_uds: |
| 429 | conn = UnixSocketHTTPConnection(self.uds_path) |
| 430 | else: |
| 431 | if self.is_https: |
| 432 | # TODO: we should verify TLS cert. |
| 433 | context = ssl.create_default_context() |
| 434 | context.check_hostname = False |
| 435 | context.verify_mode = ssl.CERT_NONE |
| 436 | conn = http.client.HTTPSConnection(self.host, context=context) |
| 437 | else: |
| 438 | conn = http.client.HTTPConnection(self.host) |
| 439 | |
| 440 | try: |
| 441 | # Make the request |
| 442 | conn.request(method, path, body=body, headers=headers) |
| 443 | response = conn.getresponse() |
| 444 | |
| 445 | status = response.status |
| 446 |
no test coverage detected