(self, api_url, req_args)
| 244 | # ================================================================= |
| 245 | |
| 246 | def _sync_send(self, api_url, req_args) -> SlackResponse: |
| 247 | params = req_args["params"] if "params" in req_args else None |
| 248 | data = req_args["data"] if "data" in req_args else None |
| 249 | files = req_args["files"] if "files" in req_args else None |
| 250 | _json = req_args["json"] if "json" in req_args else None |
| 251 | headers = req_args["headers"] if "headers" in req_args else None |
| 252 | token = params.get("token") if params and "token" in params else None |
| 253 | auth = req_args["auth"] if "auth" in req_args else None # Basic Auth for oauth.v2.access / oauth.access |
| 254 | if auth is not None: |
| 255 | headers = {} |
| 256 | if isinstance(auth, BasicAuth): |
| 257 | headers["Authorization"] = auth.encode() |
| 258 | elif isinstance(auth, str): |
| 259 | headers["Authorization"] = auth |
| 260 | else: |
| 261 | self._logger.warning(f"As the auth: {auth}: {type(auth)} is unsupported, skipped") |
| 262 | |
| 263 | body_params = {} |
| 264 | if params: |
| 265 | body_params.update(params) |
| 266 | if data: |
| 267 | body_params.update(data) |
| 268 | |
| 269 | return self._urllib_api_call( |
| 270 | token=token, |
| 271 | url=api_url, |
| 272 | query_params={}, |
| 273 | body_params=body_params, |
| 274 | files=files, |
| 275 | json_body=_json, |
| 276 | additional_headers=headers, |
| 277 | ) |
| 278 | |
| 279 | def _request_for_pagination(self, api_url: str, req_args: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: |
| 280 | """This method is supposed to be used only for SlackResponse pagination |
no test coverage detected