diff --git a/pyproject.toml b/pyproject.toml index a530cc0..3726fb0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "oxidetwitch" -version = "1.1.0" +version = "1.2.0" description = "Client for Twitch API" readme = "README.md" authors = [{ name = "Miwory", email = "miwory.uwu@gmail.com" }] @@ -99,6 +99,3 @@ ignore-one-line-docstrings = true quote-style = "single" indent-style = "space" docstring-code-format = true - -[tool.uv.sources] -aiohttpx = { index = "Miwory" } diff --git a/src/oxidetwitch/api.py b/src/oxidetwitch/api.py index d8c0966..2448d53 100644 --- a/src/oxidetwitch/api.py +++ b/src/oxidetwitch/api.py @@ -65,10 +65,15 @@ class TwitchAPIClient(OxideHTTP): raise s.InternalError(req.status_code, 'Internal Server Error') async def get_ad_schedule( - self, access_token: str, broadcaster_id: int + self, + access_token: str, + broadcaster_id: int, + *, + cache_ttl: int | None = None, ) -> s.AdSchedule: req = await self.get( '/channels/ads', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params={ 'broadcaster_id': broadcaster_id, @@ -119,12 +124,14 @@ class TwitchAPIClient(OxideHTTP): started_at: datetime | None = None, ended_at: datetime | None = None, first: int = 20, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.ExtensionAnalyticsData]: after = None while True: req = await self.get( '/analytics/extensions', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -171,12 +178,14 @@ class TwitchAPIClient(OxideHTTP): started_at: datetime | None = None, ended_at: datetime | None = None, first: int = 20, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.GameAnalyticsData]: after = None while True: req = await self.get( '/analytics/games', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -222,9 +231,11 @@ class TwitchAPIClient(OxideHTTP): *, started_at: datetime | None = None, user_id: int | None = None, + cache_ttl: int | None = None, ) -> s.BitsLeaderboard: req = await self.get( '/bits/leaderboard', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -249,10 +260,15 @@ class TwitchAPIClient(OxideHTTP): raise s.InternalError(req.status_code, 'Internal Server Error') async def get_cheermotes( - self, access_token: str, *, broadcaster_id: int | None = None + self, + access_token: str, + *, + broadcaster_id: int | None = None, + cache_ttl: int | None = None, ) -> s.Cheermotes: req = await self.get( '/bits/cheermotes', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict({'broadcaster_id': broadcaster_id}), ) @@ -276,12 +292,14 @@ class TwitchAPIClient(OxideHTTP): *, user_id: int | list[int] | None = None, first: int = 20, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.ExtensionTransactionsData]: after = None while True: req = await self.get( '/extensions/transactions', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -321,9 +339,12 @@ class TwitchAPIClient(OxideHTTP): self, access_token: str, broadcaster_id: int | list[int], + *, + cache_ttl: int | None = None, ) -> s.ChannelsInformation: req = await self.get( '/channels', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -390,10 +411,15 @@ class TwitchAPIClient(OxideHTTP): raise s.InternalError(req.status_code, 'Internal Server Error') async def get_channel_editors( - self, access_token: str, broadcaster_id: int + self, + access_token: str, + broadcaster_id: int, + *, + cache_ttl: int | None = None, ) -> s.ChannelEditors: req = await self.get( '/channels/editors', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -416,12 +442,14 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int, *, first: int = 20, + cache_ttl: int | None = None, ) -> AsyncGenerator[tuple[int, s.FollowedChannel]]: after = None while True: req = await self.get( '/channels/followed', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -463,12 +491,14 @@ class TwitchAPIClient(OxideHTTP): *, user_id: int | None = None, first: int = 20, + cache_ttl: int | None = None, ) -> AsyncGenerator[tuple[int, s.ChannelFollower]]: after = None while True: req = await self.get( '/channels/followers', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -601,9 +631,11 @@ class TwitchAPIClient(OxideHTTP): *, reward_id: str | None = None, only_manageable_rewards: bool = False, + cache_ttl: int | None = None, ) -> s.CustomRewards: req = await self.get( '/channel_points/custom_rewards', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -642,12 +674,14 @@ class TwitchAPIClient(OxideHTTP): redemption_id: int | list[int] | None = None, sort: Literal['OLDEST', 'NEWEST'] = 'OLDEST', first: int = 20, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.CustomRewardRedemption]: after = None while True: req = await self.get( '/channel_points/custom_rewards/redemptions', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -797,10 +831,15 @@ class TwitchAPIClient(OxideHTTP): raise s.InternalError(req.status_code, 'Internal Server Error') async def get_charity_campaign( - self, access_token: str, broadcaster_id: int + self, + access_token: str, + broadcaster_id: int, + *, + cache_ttl: int | None = None, ) -> s.CharityCampaign: req = await self.get( '/charity/campaigns', + cache_ttl=cache_ttl, params={'broadcaster_id': broadcaster_id}, headers={'Authorization': f'Bearer {access_token}'}, ) @@ -823,13 +862,15 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int, *, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.CharityDonation]: after = None while True: req = await self.get( '/charity/donations', + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -837,12 +878,6 @@ class TwitchAPIClient(OxideHTTP): 'after': after, } ), - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), ) match req.status_code: @@ -876,13 +911,15 @@ class TwitchAPIClient(OxideHTTP): moderator_id: int, *, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[tuple[int, s.ChattersData]]: after = None while True: req = await self.get( '/chat/chatters', + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -891,12 +928,6 @@ class TwitchAPIClient(OxideHTTP): 'after': after, } ), - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), ) match req.status_code: @@ -928,16 +959,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, broadcaster_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.ChannelEmotes: req = await self.get( '/chat/emotes', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -954,16 +981,12 @@ class TwitchAPIClient(OxideHTTP): raise s.InternalError(req.status_code, 'Internal Server Error') async def get_global_emotes( - self, access_token: str, *, cache_time: int | None = None + self, access_token: str, *, cache_ttl: int | None = None ) -> s.GlobalEmotes: req = await self.get( '/chat/emotes/global', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, ) match req.status_code: @@ -983,16 +1006,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, emote_set_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.EmoteSets: req = await self.get( '/chat/emotes/set', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'emote_set_id': emote_set_id}, ) @@ -1013,16 +1032,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, broadcaster_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.ChannelChatBadges: req = await self.get( '/chat/badges', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -1039,16 +1054,12 @@ class TwitchAPIClient(OxideHTTP): raise s.InternalError(req.status_code, 'Internal Server Error') async def get_global_chat_badges( - self, access_token: str, *, cache_time: int | None = None + self, access_token: str, *, cache_ttl: int | None = None ) -> s.GlobalChatBadges: req = await self.get( '/chat/badges/global', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, ) match req.status_code: @@ -1069,16 +1080,12 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int, *, moderator_id: int | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.ChatSettings: req = await self.get( '/chat/settings', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -1104,16 +1111,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, broadcaster_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.SharedChatSession: req = await self.get( '/shared_chat/session', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -1135,19 +1138,15 @@ class TwitchAPIClient(OxideHTTP): user_id: int, *, broadcaster_id: int | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[tuple[str, s.ChannelEmote]]: after = None while True: req = await self.get( '/chat/emotes/global', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'user_id': user_id, @@ -1345,16 +1344,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, user_id: int | list[int], *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.UserChatColor: req = await self.get( '/chat/colors', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'user_id': user_id}, ) @@ -1435,19 +1430,15 @@ class TwitchAPIClient(OxideHTTP): first: int = 20, before: str | None = None, is_featured: bool | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Clip]: after = None while True: req = await self.get( '/clips', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -1494,16 +1485,12 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int, clip_id: str, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.ClipsDownloads: req = await self.get( '/clips/downloads', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'editor_id': editor_id, @@ -1531,16 +1518,12 @@ class TwitchAPIClient(OxideHTTP): raise s.InternalError(req.status_code, 'Internal Server Error') async def get_conduits( - self, access_token: str, *, cache_time: int | None = None + self, access_token: str, *, cache_ttl: int | None = None ) -> s.Conduits: req = await self.get( '/eventsub/conduits', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, ) match req.status_code: @@ -1622,19 +1605,15 @@ class TwitchAPIClient(OxideHTTP): conduit_id: str, *, status: str | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.ConduitShard]: after = None while True: req = await self.get( '/eventsub/conduits/shards', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( {'id': conduit_id, 'status': status, 'after': after} ), @@ -1722,9 +1701,12 @@ class TwitchAPIClient(OxideHTTP): 'zh-CN', 'zh-TW', ] = 'en-US', + *, + cache_ttl: int | None = None, ) -> s.ContentClassificationLabels: req = await self.get( '/content_classification_labels', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params={'locale': locale}, ) @@ -1752,19 +1734,15 @@ class TwitchAPIClient(OxideHTTP): game_id: int | None = None, fulfillment_status: Literal['CLAIMED', 'FULFILLED'] | None = None, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.DropEntitlement]: after = None while True: req = await self.get( '/entitlements/drops', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'drop_id': drop_id, @@ -1845,16 +1823,12 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int | None = None, extension_id: int | None = None, segment: Literal['broadcaster', 'developer', 'global'] | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.ExtensionConfigurationSegment: req = await self.get( '/extensions/configurations', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {jwt_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {jwt_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -1985,19 +1959,15 @@ class TwitchAPIClient(OxideHTTP): extension_id: str, *, first: int | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.ExtensionLiveChannel]: after = None while True: req = await self.get( '/extensions/live', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'extension_id': extension_id, @@ -2036,16 +2006,12 @@ class TwitchAPIClient(OxideHTTP): jwt_token: str, extension_id: str, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.ExtensionSecrets: req = await self.get( '/extensions/jwt/secrets', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {jwt_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {jwt_token}'}, params={'extension_id': extension_id}, ) @@ -2119,16 +2085,12 @@ class TwitchAPIClient(OxideHTTP): extension_id: str, *, extension_version: str | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.Extensions: req = await self.get( '/extensions', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {jwt_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {jwt_token}'}, params=self.clean_dict( { 'extension_id': extension_id, @@ -2155,16 +2117,12 @@ class TwitchAPIClient(OxideHTTP): extension_id: str, *, extension_version: str | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.Extensions: req = await self.get( '/extensions/released', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'extension_id': extension_id, @@ -2190,16 +2148,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, *, should_include_all: bool | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.ExtensionBitsProducts: req = await self.get( '/bits/extensions', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'should_include_all': should_include_all, @@ -2341,19 +2295,15 @@ class TwitchAPIClient(OxideHTTP): sub_type: sub_type.Any | None = None, user_id: int | None = None, subscription_id: str | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[tuple[int, sub.Any]]: after = None while True: req = await self.get( '/eventsub/subscriptions', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'status': status, @@ -2397,19 +2347,15 @@ class TwitchAPIClient(OxideHTTP): *, first: int = 20, before: str | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Game]: after = None while True: req = await self.get( '/games/top', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'first': first, @@ -2450,16 +2396,12 @@ class TwitchAPIClient(OxideHTTP): name: str, *, igdb_id: str | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.Games: req = await self.get( '/games', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'id': game_id, @@ -2486,16 +2428,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, broadcaster_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.CreatorGoals: req = await self.get( '/creator_goals', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -2518,16 +2456,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, broadcaster_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.HypeTrainStatus: req = await self.get( '/hypetrain/status', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -2548,16 +2482,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, broadcaster_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.AutomodStatus: req = await self.get( '/moderation/enforcements/status', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -2611,16 +2541,12 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int, moderator_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.AutomodSettings: req = await self.get( '/moderation/automod/settings', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={ 'broadcaster_id': broadcaster_id, 'moderator_id': moderator_id, @@ -2694,13 +2620,14 @@ class TwitchAPIClient(OxideHTTP): user_id: int | list[int], *, first: int = 20, - before: str | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.BannedUser]: after = None while True: req = await self.get( '/moderation/banned', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -2708,7 +2635,6 @@ class TwitchAPIClient(OxideHTTP): 'user_id': user_id, 'first': first, 'after': after, - 'before': before, } ), ) @@ -2827,12 +2753,14 @@ class TwitchAPIClient(OxideHTTP): *, user_id: int | None = None, first: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.UnbanRequest]: after = None while True: req = await self.get( '/moderation/unbans', + cache_ttl=cache_ttl, headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { @@ -2913,19 +2841,15 @@ class TwitchAPIClient(OxideHTTP): moderator_id: int, *, first: int | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.BlockedTerm]: after = None while True: req = await self.get( '/moderation/blocked_terms', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -3058,19 +2982,15 @@ class TwitchAPIClient(OxideHTTP): user_id: int, *, first: int | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.ModeratedChannel]: after = None while True: req = await self.get( '/moderation/channels', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'user_id': user_id, @@ -3113,19 +3033,15 @@ class TwitchAPIClient(OxideHTTP): *, user_id: int | list[int] | None = None, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Moderator]: after = None while True: req = await self.get( '/moderation/moderators', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -3214,19 +3130,15 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int, *, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.VIP]: after = None while True: req = await self.get( '/channels/vips', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'user_id': user_id, @@ -3354,16 +3266,12 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int, moderator_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.ShieldModeStatus: req = await self.get( '/moderation/shield_mode', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={ 'broadcaster_id': broadcaster_id, 'moderator_id': moderator_id, @@ -3431,19 +3339,15 @@ class TwitchAPIClient(OxideHTTP): *, poll_id: int | list[int] | None = None, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Poll]: after = None while True: req = await self.get( '/polls', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -3555,19 +3459,15 @@ class TwitchAPIClient(OxideHTTP): *, prediction_id: int | list[int] | None = None, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Prediction]: after = None while True: req = await self.get( '/predictions', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -3737,19 +3637,15 @@ class TwitchAPIClient(OxideHTTP): segment_id: int | list[int] | None = None, start_time: datetime | None = None, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Schedule]: after = None while True: req = await self.get( '/schedule', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -3795,16 +3691,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, broadcaster_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> bytes: req = await self.get( '/schedule/icalendar', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -3975,19 +3867,15 @@ class TwitchAPIClient(OxideHTTP): query: str, *, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Category]: after = None while True: req = await self.get( '/search/categories', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( {'query': query, 'first': first, 'after': after} ), @@ -4024,19 +3912,15 @@ class TwitchAPIClient(OxideHTTP): *, live_only: bool = False, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Channel]: after = None while True: req = await self.get( '/search/channels', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'query': query, @@ -4076,16 +3960,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, broadcaster_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.StreamKeys: req = await self.get( '/streams/key', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -4112,19 +3992,15 @@ class TwitchAPIClient(OxideHTTP): language: str | None = None, first: int = 20, before: str | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Stream]: after = None while True: req = await self.get( '/streams', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'user_id': user_id, @@ -4169,19 +4045,15 @@ class TwitchAPIClient(OxideHTTP): user_id: int, *, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Stream]: after = None while True: req = await self.get( '/streams/followed', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'user_id': user_id, @@ -4257,19 +4129,15 @@ class TwitchAPIClient(OxideHTTP): *, first: int = 20, before: str | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.StreamMarkersData]: after = None while True: req = await self.get( '/streams/markers', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'user_id': user_id, @@ -4317,19 +4185,15 @@ class TwitchAPIClient(OxideHTTP): *, first: int = 20, before: str | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[tuple[int, int, s.BroadcasterSubscription]]: after = None while True: req = await self.get( '/subscriptions', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -4376,16 +4240,12 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int, user_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.UserSubscription: req = await self.get( '/subscriptions/user', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -4411,16 +4271,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, broadcaster_id: int, *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.ChannelTeams: req = await self.get( '/teams/channel', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'broadcaster_id': broadcaster_id}, ) @@ -4442,16 +4298,12 @@ class TwitchAPIClient(OxideHTTP): *, name: str | None = None, team_id: int | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.Teams: req = await self.get( '/teams', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict({'name': name, 'id': team_id}), ) @@ -4473,16 +4325,12 @@ class TwitchAPIClient(OxideHTTP): *, user_id: int | list[int] | None = None, login: str | list[str] | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.Users: req = await self.get( '/users', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict({'id': user_id, 'login': login}), ) @@ -4528,16 +4376,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, user_id: int | list[int], *, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.AuthorizationsByUser: req = await self.get( '/authorization/users', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'user_id': user_id}, ) @@ -4564,19 +4408,15 @@ class TwitchAPIClient(OxideHTTP): broadcaster_id: int, *, first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.UserBlock]: after = None while True: req = await self.get( '/users/blocks', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'broadcaster_id': broadcaster_id, @@ -4672,16 +4512,12 @@ class TwitchAPIClient(OxideHTTP): raise s.InternalError(req.status_code, 'Internal Server Error') async def get_user_extensions( - self, access_token: str, *, cache_time: int | None = None + self, access_token: str, *, cache_ttl: int | None = None ) -> s.UserExtensions: req = await self.get( '/users/extensions/list', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, ) match req.status_code: @@ -4701,16 +4537,12 @@ class TwitchAPIClient(OxideHTTP): access_token: str, *, user_id: int | None = None, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> s.UserActiveExtensions: req = await self.get( '/users/extensions', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params={'user_id': user_id}, ) @@ -4761,19 +4593,15 @@ class TwitchAPIClient(OxideHTTP): sort: Literal['time', 'trending', 'views'] = 'time', video_type: Literal['all', 'archive', 'highlight', 'upload'] = 'all', first: int = 20, - cache_time: int | None = None, + cache_ttl: int | None = None, ) -> AsyncGenerator[s.Video]: after = None while True: req = await self.get( '/videos', - headers=self.clean_dict( - { - 'Authorization': f'Bearer {access_token}', - 'X-Cache-TTL': str(cache_time), - } - ), + cache_ttl=cache_ttl, + headers={'Authorization': f'Bearer {access_token}'}, params=self.clean_dict( { 'video_id': video_id,