| 171 | |
| 172 | |
| 173 | def http_req( |
| 174 | hostname: str = "", |
| 175 | port: str = "443", |
| 176 | method: Optional[str] = None, |
| 177 | headers: MutableMapping[str, str] = {}, |
| 178 | data: Optional[str] = None, |
| 179 | endpoint: str = "/", |
| 180 | scheme: str = "https", |
| 181 | ssl_verify: bool = False, |
| 182 | timeout: Optional[int] = None, |
| 183 | ssl_ctx: Optional[Any] = None, |
| 184 | ) -> Tuple[Any, Any, Any]: |
| 185 | |
| 186 | if not ssl_ctx: |
| 187 | ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
| 188 | if not ssl_verify: |
| 189 | ssl_ctx.check_hostname = False |
| 190 | ssl_ctx.verify_mode = ssl.CERT_NONE |
| 191 | else: |
| 192 | ssl_ctx.verify_mode = ssl.CERT_REQUIRED |
| 193 | |
| 194 | url: str = f"{scheme}://{hostname}:{port}{endpoint}" |
| 195 | _data = bytes(data, "ascii") if data else None |
| 196 | _headers = headers |
| 197 | if data and not method: |
| 198 | method = "POST" |
| 199 | if not _headers.get("Content-Type") and method in ["POST", "PATCH"]: |
| 200 | _headers["Content-Type"] = "application/json" |
| 201 | try: |
| 202 | req = Request(url, _data, _headers, method=method) |
| 203 | with urlopen(req, context=ssl_ctx, timeout=timeout) as response: |
| 204 | response_str = response.read() |
| 205 | response_headers = response.headers |
| 206 | response_code = response.code |
| 207 | return response_headers, response_str.decode(), response_code |
| 208 | except (HTTPError, URLError) as e: |
| 209 | # Log level is debug only. |
| 210 | # We let whatever calls `http_req()` catching and printing the error |
| 211 | logger.debug(f"url={url} err={e}") |
| 212 | # handle error here if needed |
| 213 | raise |
| 214 | |
| 215 | |
| 216 | def write_tmp_file( |