(self, api_url, req_args)
| 207 | # ================================================================= |
| 208 | |
| 209 | def _sync_send(self, api_url, req_args) -> SlackResponse: |
| 210 | params = req_args["params"] if "params" in req_args else None |
| 211 | data = req_args["data"] if "data" in req_args else None |
| 212 | files = req_args["files"] if "files" in req_args else None |
| 213 | _json = req_args["json"] if "json" in req_args else None |
| 214 | headers = req_args["headers"] if "headers" in req_args else None |
| 215 | token = params.get("token") if params and "token" in params else None |
| 216 | auth = req_args["auth"] if "auth" in req_args else None # Basic Auth for oauth.v2.access / oauth.access |
| 217 | if auth is not None: |
| 218 | if isinstance(auth, BasicAuth): |
| 219 | headers["Authorization"] = auth.encode() |
| 220 | elif isinstance(auth, str): |
| 221 | headers["Authorization"] = auth |
| 222 | else: |
| 223 | self._logger.warning(f"As the auth: {auth}: {type(auth)} is unsupported, skipped") |
| 224 | |
| 225 | body_params = {} |
| 226 | if params: |
| 227 | body_params.update(params) |
| 228 | if data: |
| 229 | body_params.update(data) |
| 230 | |
| 231 | return self._urllib_api_call( |
| 232 | token=token, |
| 233 | url=api_url, |
| 234 | query_params={}, |
| 235 | body_params=body_params, |
| 236 | files=files, |
| 237 | json_body=_json, |
| 238 | additional_headers=headers, |
| 239 | ) |
| 240 | |
| 241 | def _request_for_pagination(self, api_url, req_args) -> Dict[str, any]: |
| 242 | """This method is supposed to be used only for SlackResponse pagination |
no test coverage detected