(
self,
*,
token: str = None,
url: str,
query_params: Dict[str, str] = {},
json_body: Dict = {},
body_params: Dict[str, str] = {},
files: Dict[str, io.BytesIO] = {},
additional_headers: Dict[str, str] = {},
)
| 254 | } |
| 255 | |
| 256 | def _urllib_api_call( |
| 257 | self, |
| 258 | *, |
| 259 | token: str = None, |
| 260 | url: str, |
| 261 | query_params: Dict[str, str] = {}, |
| 262 | json_body: Dict = {}, |
| 263 | body_params: Dict[str, str] = {}, |
| 264 | files: Dict[str, io.BytesIO] = {}, |
| 265 | additional_headers: Dict[str, str] = {}, |
| 266 | ) -> SlackResponse: |
| 267 | files_to_close: List[BinaryIO] = [] |
| 268 | try: |
| 269 | # True/False -> "1"/"0" |
| 270 | query_params = convert_bool_to_0_or_1(query_params) |
| 271 | body_params = convert_bool_to_0_or_1(body_params) |
| 272 | |
| 273 | if self._logger.level <= logging.DEBUG: |
| 274 | |
| 275 | def convert_params(values: dict) -> dict: |
| 276 | if not values or not isinstance(values, dict): |
| 277 | return {} |
| 278 | return {k: ("(bytes)" if isinstance(v, bytes) else v) for k, v in values.items()} |
| 279 | |
| 280 | headers = {k: "(redacted)" if k.lower() == "authorization" else v for k, v in additional_headers.items()} |
| 281 | self._logger.debug( |
| 282 | f"Sending a request - url: {url}, " |
| 283 | f"query_params: {convert_params(query_params)}, " |
| 284 | f"body_params: {convert_params(body_params)}, " |
| 285 | f"files: {convert_params(files)}, " |
| 286 | f"json_body: {json_body}, " |
| 287 | f"headers: {headers}" |
| 288 | ) |
| 289 | |
| 290 | request_data = {} |
| 291 | if files is not None and isinstance(files, dict) and len(files) > 0: |
| 292 | if body_params: |
| 293 | for k, v in body_params.items(): |
| 294 | request_data.update({k: v}) |
| 295 | |
| 296 | for k, v in files.items(): |
| 297 | if isinstance(v, str): |
| 298 | f: BinaryIO = open(v.encode("utf-8", "ignore"), "rb") |
| 299 | files_to_close.append(f) |
| 300 | request_data.update({k: f}) |
| 301 | elif isinstance(v, (bytearray, bytes)): |
| 302 | request_data.update({k: io.BytesIO(v)}) |
| 303 | else: |
| 304 | request_data.update({k: v}) |
| 305 | |
| 306 | request_headers = self._build_urllib_request_headers( |
| 307 | token=token or self.token, |
| 308 | has_json=json is not None, |
| 309 | has_files=files is not None, |
| 310 | additional_headers=additional_headers, |
| 311 | ) |
| 312 | request_args = { |
| 313 | "headers": request_headers, |
no test coverage detected