(self, path, api: Any = 'public', method='GET', params={}, headers: dict = None, body: Any = None)
| 2208 | return self.milliseconds() |
| 2209 | |
| 2210 | def sign(self, path, api: Any = 'public', method='GET', params={}, headers: dict = None, body: Any = None): |
| 2211 | url = self.implode_params(self.urls['api'][api], { |
| 2212 | 'hostname': self.hostname, |
| 2213 | }) |
| 2214 | url += '/' + self.version + '/' + self.implode_params(path, params) |
| 2215 | query = self.omit(params, self.extract_params(path)) |
| 2216 | if method != 'POST': |
| 2217 | if query: |
| 2218 | url += '?' + self.urlencode(query) |
| 2219 | if api == 'private': |
| 2220 | self.check_required_credentials() |
| 2221 | headers = {} |
| 2222 | nonce = self.uuid() |
| 2223 | request = { |
| 2224 | 'access_key': self.apiKey, |
| 2225 | 'nonce': nonce, |
| 2226 | } |
| 2227 | hasQuery = query |
| 2228 | auth = None |
| 2229 | if (method != 'GET') and (method != 'DELETE'): |
| 2230 | body = self.json(params) |
| 2231 | headers['Content-Type'] = 'application/json' |
| 2232 | if hasQuery: |
| 2233 | auth = self.rawencode(query) |
| 2234 | if auth is not None: |
| 2235 | hash = self.hash(self.encode(auth), 'sha512') |
| 2236 | request['query_hash'] = hash |
| 2237 | request['query_hash_alg'] = 'SHA512' |
| 2238 | token = self.jwt(request, self.encode(self.secret), 'sha256') |
| 2239 | headers['Authorization'] = 'Bearer ' + token |
| 2240 | return {'url': url, 'method': method, 'body': body, 'headers': headers} |
| 2241 | |
| 2242 | def handle_errors(self, httpCode: int, reason: str, url: str, method: str, headers: dict, body: str, response, requestHeaders, requestBody): |
| 2243 | if response is None: |
nothing calls this directly
no test coverage detected