This commit is contained in:
747
src/twitchclient/api.py
Normal file
747
src/twitchclient/api.py
Normal file
@ -0,0 +1,747 @@
|
||||
from datetime import datetime
|
||||
from typing import Literal
|
||||
|
||||
from aiohttpx import status as st
|
||||
from aiohttpx.client import AioHTTPXClient
|
||||
|
||||
from . import schema as s
|
||||
|
||||
|
||||
class TwitchAPIClient(AioHTTPXClient):
|
||||
def __init__(
|
||||
self,
|
||||
redis_url: str,
|
||||
client_id: str,
|
||||
client_secret: str,
|
||||
redirect_uri: str,
|
||||
):
|
||||
self.base_uri = 'https://api.twitch.tv/helix'
|
||||
self.client_id = client_id
|
||||
self.client_secret = client_secret
|
||||
self.redirect_uri = redirect_uri
|
||||
|
||||
super().__init__(
|
||||
base_url=self.base_uri,
|
||||
headers={'Client-Id': self.client_id},
|
||||
redis_url=redis_url,
|
||||
key='twitch',
|
||||
limit=10,
|
||||
logger='Twitch API',
|
||||
)
|
||||
|
||||
async def start_commercial(self, access_token: str, broadcaster_id: int):
|
||||
req = await self.get(
|
||||
'/channels/commercial',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params={
|
||||
'broadcaster_id': broadcaster_id,
|
||||
},
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.StartCommercial.model_validate(req.json()).data
|
||||
|
||||
case (
|
||||
st.BAD_REQUEST
|
||||
| st.UNAUTHORIZED
|
||||
| st.NOT_FOUND
|
||||
| st.TOO_MANY_REQUESTS
|
||||
):
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_ad_schedule(self, access_token: str, broadcaster_id: int):
|
||||
req = await self.get(
|
||||
'/channels/ads',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params={
|
||||
'broadcaster_id': broadcaster_id,
|
||||
},
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.AdSchedule.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def snooze_next_ad(self, access_token: str, broadcaster_id: int):
|
||||
req = await self.post(
|
||||
'/channels/ads/schedule/snooze',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params={
|
||||
'broadcaster_id': broadcaster_id,
|
||||
},
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.SnoozeNextAd.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.TOO_MANY_REQUESTS:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_extension_analytics(
|
||||
self,
|
||||
access_token: str,
|
||||
extension_id: str | None = None,
|
||||
analytics_type: Literal['overview_v2'] | None = None,
|
||||
started_at: datetime | None = None,
|
||||
ended_at: datetime | None = None,
|
||||
first: int = 20,
|
||||
after: str | None = None,
|
||||
):
|
||||
req = await self.get(
|
||||
'/analytics/extensions',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'extension_id': extension_id,
|
||||
'type': analytics_type,
|
||||
'started_at': started_at,
|
||||
'ended_at': ended_at,
|
||||
'first': first,
|
||||
'after': after,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.ExtensionAnalytics.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.NOT_FOUND:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_game_analytics(
|
||||
self,
|
||||
access_token: str,
|
||||
game_id: int | None = None,
|
||||
analytics_type: Literal['overview_v2'] | None = None,
|
||||
started_at: datetime | None = None,
|
||||
ended_at: datetime | None = None,
|
||||
first: int = 20,
|
||||
after: str | None = None,
|
||||
):
|
||||
req = await self.get(
|
||||
'/analytics/games',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'game_id': game_id,
|
||||
'type': analytics_type,
|
||||
'started_at': started_at,
|
||||
'ended_at': ended_at,
|
||||
'first': first,
|
||||
'after': after,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.GameAnalytics.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.NOT_FOUND:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_bits_leaderboard(
|
||||
self,
|
||||
access_token: str,
|
||||
count: int = 10,
|
||||
period: Literal['day', 'week', 'month', 'year', 'all'] = 'all',
|
||||
started_at: datetime | None = None,
|
||||
user_id: int | None = None,
|
||||
):
|
||||
req = await self.get(
|
||||
'/bits/leaderboard',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'count': count,
|
||||
'period': period,
|
||||
'started_at': started_at,
|
||||
'user_id': user_id,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.BitsLeaderboard.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.FORBIDDEN:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_cheermotes(
|
||||
self, access_token: str, broadcaster_id: int | None = None
|
||||
):
|
||||
req = await self.get(
|
||||
'/bits/cheermotes',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params=self.clean_dict({'broadcaster_id': broadcaster_id}),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.Cheermotes.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_extension_transactions(
|
||||
self,
|
||||
access_token: str,
|
||||
extension_id: str,
|
||||
user_id: int | list[int] | None = None,
|
||||
first: int = 20,
|
||||
after: str | None = None,
|
||||
):
|
||||
req = await self.get(
|
||||
'/extensions/transactions',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'extension_id': extension_id,
|
||||
'user_id': user_id,
|
||||
'first': first,
|
||||
'after': after,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.ExtensionTransactions.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.NOT_FOUND:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_channel_information(
|
||||
self, access_token: str, broadcaster_id: int | list[int]
|
||||
):
|
||||
req = await self.get(
|
||||
'/channels',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params={'broadcaster_id': broadcaster_id},
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.ChannelsInformation.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.TOO_MANY_REQUESTS:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def modify_channel_information(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
*,
|
||||
game_id: int | None = None,
|
||||
broadcaster_language: str | None = None,
|
||||
title: str | None = None,
|
||||
delay: int | None = None,
|
||||
tags: list[str] | None = None,
|
||||
content_classification_labels: list[s.ContentClassificationLabel]
|
||||
| None = None,
|
||||
is_branded_content: bool | None = None,
|
||||
):
|
||||
data = self.clean_dict(
|
||||
{
|
||||
'game_id': game_id,
|
||||
'broadcaster_language': broadcaster_language,
|
||||
'title': title,
|
||||
'delay': delay,
|
||||
'tags': tags,
|
||||
'content_classification_labels': content_classification_labels,
|
||||
'is_branded_content': is_branded_content,
|
||||
}
|
||||
)
|
||||
req = await self.patch(
|
||||
'/channels',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params={'broadcaster_id': broadcaster_id},
|
||||
json=data,
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.NO_CONTENT:
|
||||
return True
|
||||
|
||||
case (
|
||||
st.BAD_REQUEST
|
||||
| st.UNAUTHORIZED
|
||||
| st.FORBIDDEN
|
||||
| st.TOO_MANY_REQUESTS
|
||||
):
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_channel_editors(
|
||||
self, access_token: str, broadcaster_id: int
|
||||
):
|
||||
req = await self.get(
|
||||
'/channels/editors',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params={'broadcaster_id': broadcaster_id},
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.ChannelEditors.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.TOO_MANY_REQUESTS:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_followed_channels(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
*,
|
||||
first: int = 20,
|
||||
after: str | None = None,
|
||||
):
|
||||
req = await self.get(
|
||||
'/channels/followed',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'broadcaster_id': broadcaster_id,
|
||||
'first': first,
|
||||
'after': after,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.FollowedChannels.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.TOO_MANY_REQUESTS:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_channel_followers(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
*,
|
||||
user_id: int | None = None,
|
||||
first: int = 20,
|
||||
after: str | None = None,
|
||||
):
|
||||
req = await self.get(
|
||||
'/channels/followers',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'broadcaster_id': broadcaster_id,
|
||||
'user_id': user_id,
|
||||
'first': first,
|
||||
'after': after,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.ChannelFollowers.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.TOO_MANY_REQUESTS:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def create_custom_rewards(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
title: str,
|
||||
cost: int,
|
||||
*,
|
||||
prompt: str | None = None,
|
||||
is_enabled: bool = True,
|
||||
background_color: str | None = None,
|
||||
is_user_input_required: bool = False,
|
||||
is_max_per_stream_enabled: bool = False,
|
||||
max_per_stream: int | None = None,
|
||||
is_max_per_user_per_stream_enabled: bool = False,
|
||||
max_per_user_per_stream: int | None = None,
|
||||
is_global_cooldown_enabled: bool = False,
|
||||
global_cooldown_seconds: int | None = None,
|
||||
should_redemptions_skip_request_queue: bool = False,
|
||||
):
|
||||
data = self.clean_dict(
|
||||
{
|
||||
'broadcaster_id': broadcaster_id,
|
||||
'title': title,
|
||||
'cost': cost,
|
||||
'prompt': prompt,
|
||||
'is_enabled': is_enabled,
|
||||
'background_color': background_color,
|
||||
'is_user_input_required': is_user_input_required,
|
||||
'is_max_per_stream_enabled': is_max_per_stream_enabled,
|
||||
'max_per_stream': max_per_stream,
|
||||
'is_max_per_user_per_'
|
||||
'stream_enabled': is_max_per_user_per_stream_enabled,
|
||||
'max_per_user_per_stream': max_per_user_per_stream,
|
||||
'is_global_cooldown_enabled': is_global_cooldown_enabled,
|
||||
'global_cooldown_seconds': global_cooldown_seconds,
|
||||
'should_redemptions_skip_'
|
||||
'request_queue': should_redemptions_skip_request_queue,
|
||||
}
|
||||
)
|
||||
|
||||
req = await self.post(
|
||||
'/channel_points/custom_rewards',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
json=data,
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.CustomRewards.model_validate(req.json()).data
|
||||
|
||||
case (
|
||||
st.BAD_REQUEST
|
||||
| st.UNAUTHORIZED
|
||||
| st.FORBIDDEN
|
||||
| st.TOO_MANY_REQUESTS
|
||||
):
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def delete_custom_reward(
|
||||
self, access_token: str, broadcaster_id: str, reward_id: str
|
||||
):
|
||||
req = await self.delete(
|
||||
'/channel_points/custom_rewards',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params={'broadcaster_id': broadcaster_id, 'id': reward_id},
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.NO_CONTENT:
|
||||
return True
|
||||
|
||||
case (
|
||||
st.BAD_REQUEST
|
||||
| st.UNAUTHORIZED
|
||||
| st.FORBIDDEN
|
||||
| st.NOT_FOUND
|
||||
| st.TOO_MANY_REQUESTS
|
||||
):
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_custom_rewards(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
*,
|
||||
reward_id: str | None = None,
|
||||
only_manageable_rewards: bool = False,
|
||||
):
|
||||
req = await self.get(
|
||||
'/channel_points/custom_rewards',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'broadcaster_id': broadcaster_id,
|
||||
'id': reward_id,
|
||||
'only_manageable_rewards': only_manageable_rewards,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.CustomRewards.model_validate(req.json()).data
|
||||
|
||||
case (
|
||||
st.BAD_REQUEST
|
||||
| st.UNAUTHORIZED
|
||||
| st.FORBIDDEN
|
||||
| st.NOT_FOUND
|
||||
| st.TOO_MANY_REQUESTS
|
||||
):
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_custom_reward_redemption(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
reward_id: str,
|
||||
status: Literal['CANCELED', 'FULFILLED', 'UNFULFILLED'],
|
||||
*,
|
||||
redemption_id: int | list[int] | None = None,
|
||||
sort: Literal['OLDEST', 'NEWEST'] = 'OLDEST',
|
||||
after: str | None = None,
|
||||
first: int = 20,
|
||||
):
|
||||
req = await self.get(
|
||||
'/channel_points/custom_rewards/redemptions',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'broadcaster_id': broadcaster_id,
|
||||
'reward_id': reward_id,
|
||||
'status': status,
|
||||
'redemption_id': redemption_id,
|
||||
'sort': sort,
|
||||
'after': after,
|
||||
'first': first,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.CustomRewardRedemptions.model_validate(
|
||||
req.json()
|
||||
).data
|
||||
|
||||
case (
|
||||
st.BAD_REQUEST
|
||||
| st.UNAUTHORIZED
|
||||
| st.NOT_FOUND
|
||||
| st.TOO_MANY_REQUESTS
|
||||
):
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def update_custom_reward(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
reward_id: str,
|
||||
*,
|
||||
title: str | None = None,
|
||||
prompt: str | None = None,
|
||||
cost: int | None = None,
|
||||
background_color: str | None = None,
|
||||
is_enabled: bool | None = None,
|
||||
is_user_input_required: bool | None = None,
|
||||
is_max_per_stream_enabled: bool | None = None,
|
||||
max_per_stream: int | None = None,
|
||||
is_max_per_user_per_stream_enabled: bool | None = None,
|
||||
max_per_user_per_stream: int | None = None,
|
||||
is_global_cooldown_enabled: bool | None = None,
|
||||
global_cooldown_seconds: int | None = None,
|
||||
is_paused: bool | None = None,
|
||||
should_redemptions_skip_request_queue: bool | None = None,
|
||||
):
|
||||
data = self.clean_dict(
|
||||
{
|
||||
'title': title,
|
||||
'prompt': prompt,
|
||||
'cost': cost,
|
||||
'background_color': background_color,
|
||||
'is_enabled': is_enabled,
|
||||
'is_user_input_required': is_user_input_required,
|
||||
'is_max_per_stream_enabled': is_max_per_stream_enabled,
|
||||
'max_per_stream': max_per_stream,
|
||||
'is_max_per_user_per_'
|
||||
'stream_enabled': is_max_per_user_per_stream_enabled,
|
||||
'max_per_user_per_stream': max_per_user_per_stream,
|
||||
'is_global_cooldown_enabled': is_global_cooldown_enabled,
|
||||
'global_cooldown_seconds': global_cooldown_seconds,
|
||||
'is_paused': is_paused,
|
||||
'should_redemptions_skip_'
|
||||
'request_queue': should_redemptions_skip_request_queue,
|
||||
}
|
||||
)
|
||||
req = await self.patch(
|
||||
'/channel_points/custom_rewards',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params={'broadcaster_id': broadcaster_id, 'id': reward_id},
|
||||
json=data,
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.CustomRewards.model_validate(req.json()).data
|
||||
|
||||
case (
|
||||
st.BAD_REQUEST
|
||||
| st.UNAUTHORIZED
|
||||
| st.FORBIDDEN
|
||||
| st.NOT_FOUND
|
||||
| st.TOO_MANY_REQUESTS
|
||||
):
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def update_redemption_status(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
reward_id: int,
|
||||
redemption_id: int | list[int],
|
||||
status: Literal['CANCELED', 'FULFILLED'],
|
||||
):
|
||||
req = await self.post(
|
||||
'/channel_points/custom_rewards/redemptions',
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
params={
|
||||
'broadcaster_id': broadcaster_id,
|
||||
'reward_id': reward_id,
|
||||
'id': redemption_id,
|
||||
},
|
||||
json={'status': status},
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.NO_CONTENT:
|
||||
return True
|
||||
|
||||
case (
|
||||
st.BAD_REQUEST
|
||||
| st.UNAUTHORIZED
|
||||
| st.FORBIDDEN
|
||||
| st.NOT_FOUND
|
||||
| st.TOO_MANY_REQUESTS
|
||||
):
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_charity_campaign(
|
||||
self, access_token: str, broadcaster_id: int
|
||||
):
|
||||
req = await self.get(
|
||||
'/charity/campaigns',
|
||||
params={'broadcaster_id': broadcaster_id},
|
||||
headers={'Authorization': f'Bearer {access_token}'},
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.CharityCampaign.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.FORBIDDEN:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_charity_campaign_donations(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
first: int = 20,
|
||||
after: str | None = None,
|
||||
cache_time: int | None = None,
|
||||
):
|
||||
req = await self.get(
|
||||
'/charity/donations',
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'broadcaster_id': broadcaster_id,
|
||||
'first': first,
|
||||
'after': after,
|
||||
}
|
||||
),
|
||||
headers=self.clean_dict(
|
||||
{
|
||||
'Authorization': f'Bearer {access_token}',
|
||||
'X-Cache-TTL': cache_time,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.CharityDonations.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.FORBIDDEN:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
|
||||
async def get_chatters(
|
||||
self,
|
||||
access_token: str,
|
||||
broadcaster_id: int,
|
||||
moderator_id: int,
|
||||
first: int = 20,
|
||||
after: str | None = None,
|
||||
cache_time: int | None = None,
|
||||
):
|
||||
req = await self.get(
|
||||
'/chatters',
|
||||
params=self.clean_dict(
|
||||
{
|
||||
'broadcaster_id': broadcaster_id,
|
||||
'moderator_id': moderator_id,
|
||||
'first': first,
|
||||
'after': after,
|
||||
}
|
||||
),
|
||||
headers=self.clean_dict(
|
||||
{
|
||||
'Authorization': f'Bearer {access_token}',
|
||||
'X-Cache-TTL': cache_time,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
match req.status_code:
|
||||
case st.OK:
|
||||
return s.Chatters.model_validate(req.json()).data
|
||||
|
||||
case st.BAD_REQUEST | st.UNAUTHORIZED | st.FORBIDDEN:
|
||||
raise s.Error(req.status_code, req.json()['message'])
|
||||
|
||||
case _:
|
||||
raise s.Error(req.status_code, 'Internal Server Error')
|
||||
Reference in New Issue
Block a user