Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
reset = resp.headers.get('Ratelimit-Reset')
remaining = resp.headers.get('Ratelimit-Remaining')
self._bucket.update(reset=reset, remaining=remaining)
if 200 <= resp.status < 300:
if resp.content_type == 'application/json':
return await resp.json(), False
return await resp.text(encoding='utf-8'), True
if resp.status == 401:
if self.client_id is None:
raise Unauthorized('A client ID or other authorization is needed to use this route.')
raise Unauthorized('You\'re not authorized to use this route.')
if resp.status == 429:
reason = 'Ratelimit Reached'
if not utilize_bucket: # non Helix APIs don't have ratelimit headers
await asyncio.sleep(3 ** attempt + 1)
continue
raise HTTPException(f'Failed to fulfil request ({resp.status}).', resp.reason)
raise HTTPException('Failed to reach Twitch API', reason)
if utilize_bucket:
reset = resp.headers.get('Ratelimit-Reset')
remaining = resp.headers.get('Ratelimit-Remaining')
self._bucket.update(reset=reset, remaining=remaining)
if 200 <= resp.status < 300:
if resp.content_type == 'application/json':
return await resp.json(), False
return await resp.text(encoding='utf-8'), True
if resp.status == 401:
if self.client_id is None:
raise Unauthorized('A client ID or other authorization is needed to use this route.')
raise Unauthorized('You\'re not authorized to use this route.')
if resp.status == 429:
reason = 'Ratelimit Reached'
if not utilize_bucket: # non Helix APIs don't have ratelimit headers
await asyncio.sleep(3 ** attempt + 1)
continue
raise HTTPException(f'Failed to fulfil request ({resp.status}).', resp.reason)
raise HTTPException('Failed to reach Twitch API', reason)