Performs a Slack API request and returns the result. Args: token: Slack API Token (either bot token or user token) url: Complete URL (e.g., https://slack.com/api/chat.postMessage) query_params: Query string json_body: JSON data structure (it's
(
self,
*,
token: Optional[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],
)
| 223 | } |
| 224 | |
| 225 | def _urllib_api_call( |
| 226 | self, |
| 227 | *, |
| 228 | token: Optional[str] = None, |
| 229 | url: str, |
| 230 | query_params: Dict[str, str], |
| 231 | json_body: Dict, |
| 232 | body_params: Dict[str, str], |
| 233 | files: Dict[str, io.BytesIO], |
| 234 | additional_headers: Dict[str, str], |
| 235 | ) -> SlackResponse: |
| 236 | """Performs a Slack API request and returns the result. |
| 237 | |
| 238 | Args: |
| 239 | token: Slack API Token (either bot token or user token) |
| 240 | url: Complete URL (e.g., https://slack.com/api/chat.postMessage) |
| 241 | query_params: Query string |
| 242 | json_body: JSON data structure (it's still a dict at this point), |
| 243 | if you give this argument, body_params and files will be skipped |
| 244 | body_params: Form body params |
| 245 | files: Files to upload |
| 246 | additional_headers: Request headers to append |
| 247 | |
| 248 | Returns: |
| 249 | API response |
| 250 | """ |
| 251 | files_to_close: List[BinaryIO] = [] |
| 252 | try: |
| 253 | # True/False -> "1"/"0" |
| 254 | query_params = convert_bool_to_0_or_1(query_params) # type: ignore[assignment] |
| 255 | body_params = convert_bool_to_0_or_1(body_params) # type: ignore[assignment] |
| 256 | |
| 257 | if self._logger.level <= logging.DEBUG: |
| 258 | |
| 259 | def convert_params(values: dict) -> dict: |
| 260 | if not values or not isinstance(values, dict): |
| 261 | return {} |
| 262 | return {k: ("(bytes)" if isinstance(v, bytes) else v) for k, v in values.items()} |
| 263 | |
| 264 | headers = {k: "(redacted)" if k.lower() == "authorization" else v for k, v in additional_headers.items()} |
| 265 | self._logger.debug( |
| 266 | f"Sending a request - url: {url}, " |
| 267 | f"query_params: {convert_params(query_params)}, " |
| 268 | f"body_params: {convert_params(body_params)}, " |
| 269 | f"files: {convert_params(files)}, " |
| 270 | f"json_body: {json_body}, " |
| 271 | f"headers: {headers}" |
| 272 | ) |
| 273 | |
| 274 | request_data = {} |
| 275 | if files is not None and isinstance(files, dict) and len(files) > 0: |
| 276 | if body_params: |
| 277 | for k, v in body_params.items(): |
| 278 | request_data.update({k: v}) |
| 279 | |
| 280 | for k, v in files.items(): # type: ignore[assignment] |
| 281 | if isinstance(v, str): |
| 282 | f: BinaryIO = open(v.encode("utf-8", "ignore"), "rb") |
no test coverage detected