Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _populate_entries(*channels: Union[str, int]):
names = set()
ids = set()
for channel in channels:
if isinstance(channel, str):
if channel.isdigit():
# Handle ids in the string form
ids.add(int(channel))
else:
names.add(channel)
elif isinstance(channel, int):
ids.add(str(channel))
if len(names | ids) > 100:
raise HTTPException('Bad Request - Total entries must not exceed 100.')
return names, ids
async def handle_callback(self, request) -> web.Response:
query = request.query
try:
if query['hub.mode'] == 'denied':
asyncio.run_coroutine_threadsafe(self._bot._ws.event_error(
HTTPException(f'Webhook subscription denied | {query["hub.reason"]}')), loop=self.loop)
return web.Response(text='200: OK', status=200)
if query['hub.challenge']:
asyncio.run_coroutine_threadsafe(self._bot.event_webhook(query), loop=self.loop)
return web.Response(body=query['hub.challenge'],
content_type='application/json')
except KeyError:
web.Response(text='Bad Request', status=400)
return web.Response(text='200: OK', status=200)
class InvalidContent(TwitchIOBException):
pass
class HTTPException(TwitchIOBException):
pass
class EchoMessageWarning(TwitchIOBException):
"""Exception raised when a bot user tries to repsond to it's own message."""
pass
class Unauthorized(HTTPException):
pass
Chatters
Namedtuple containing active chatter data.
Raises
--------
HTTPException
Bad request while fetching stream chatters.
"""
url = f'http://tmi.twitch.tv/group/user/{channel.lower()}/chatters'
async with self.http._session.get(url) as resp:
if 200 <= resp.status < 300:
data = await resp.json()
else:
raise HTTPException(f'Fetching chatters failed: {resp.status}', resp.reason)
all_ = []
for x in data['chatters'].values():
all_ += x
return Chatters(data['chatter_count'], all_, *data['chatters'].values())
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)
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)