-------------------- Generic function to call and return SxT API. This is the base api execution function. It can, but is not intended, to be used directly. Rather, it is wrapped by other api-specific functions, to isolate api call differences from the actual api e
(self, endpoint: str,
auth_header:bool = True,
request_type:str = SXTApiCallTypes.POST,
header_parms: dict = {},
data_parms: dict = {},
query_parms: dict = {},
path_parms: dict = {},
endpoint_full_override_flag: bool = False )
| 136 | |
| 137 | |
| 138 | def call_api(self, endpoint: str, |
| 139 | auth_header:bool = True, |
| 140 | request_type:str = SXTApiCallTypes.POST, |
| 141 | header_parms: dict = {}, |
| 142 | data_parms: dict = {}, |
| 143 | query_parms: dict = {}, |
| 144 | path_parms: dict = {}, |
| 145 | endpoint_full_override_flag: bool = False ) -> tuple[bool, object]: |
| 146 | """-------------------- |
| 147 | Generic function to call and return SxT API. |
| 148 | |
| 149 | This is the base api execution function. It can, but is not intended, to be used directly. |
| 150 | Rather, it is wrapped by other api-specific functions, to isolate api call differences |
| 151 | from the actual api execution, which can all be the same. |
| 152 | |
| 153 | Args: |
| 154 | endpoint (str): URL endpoint, after the version. Final structure is: [api_url/version/endpoint] |
| 155 | request_type (SXTApiCallTypes): Type of request. [POST, GET, PUT, DELETE] |
| 156 | auth_header (bool): flag indicator whether to append the Bearer token to the header. |
| 157 | header_parms: (dict): Name/Value pair to add to request header, except for bearer token. {Name: Value} |
| 158 | query_parms: (dict): Name/value pairs to be added to the query string. {Name: Value} |
| 159 | data_parms (dict): Dictionary to be used holistically for --data json object. |
| 160 | path_parms (dict): Pattern to replace placeholders in URL. {Placeholder_in_URL: Replace_Value} |
| 161 | endpoint_full_override_flag (str): If True, endpoint is used verbatium, rather than constructing version + endpoint. Will negate any querystring or path parms. |
| 162 | |
| 163 | Results: |
| 164 | bool: Indicating request success |
| 165 | json: Result of the API, expressed as a JSON object |
| 166 | """ |
| 167 | # Set these early, in case of timeout and they're not set by callfunc |
| 168 | txt = 'response.text not available - are you sure you have the correct API Endpoint?' |
| 169 | statuscode = 555 |
| 170 | response = {} |
| 171 | |
| 172 | # if network calls turned off, return fake data |
| 173 | if not self.network_calls_enabled: return True, self.__fakedata__(endpoint) |
| 174 | |
| 175 | # internal function to simplify and unify error handling |
| 176 | def __handle_errors__(txt, ex, statuscode, responseobject, loggerobject): |
| 177 | loggerobject.error(txt) |
| 178 | rtn = {'text':txt} |
| 179 | rtn['error'] = str(ex) |
| 180 | rtn['status_code'] = statuscode |
| 181 | rtn['response_object'] = responseobject |
| 182 | return False, rtn |
| 183 | |
| 184 | # otherwise, go get real data |
| 185 | try: |
| 186 | # Header parms |
| 187 | headers = {k:v for k,v in self.standard_headers.items()} # get new object |
| 188 | if auth_header: headers['authorization'] = f'Bearer {self.access_token}' |
| 189 | headers.update(header_parms) |
| 190 | |
| 191 | |
| 192 | if endpoint_full_override_flag: |
| 193 | url = endpoint |
| 194 | self.logger.debug(f'API Call started for (custom) endpoint: {endpoint}') |
| 195 |