commit e437b41ee80d4e2c14c33f844f0b5ffb0bd56665 Author: Miwory Date: Thu Dec 11 15:56:56 2025 +0300 Альфа релиз diff --git a/.gitea/workflows/latest.yaml b/.gitea/workflows/latest.yaml new file mode 100644 index 0000000..59305ac --- /dev/null +++ b/.gitea/workflows/latest.yaml @@ -0,0 +1,64 @@ +name: Build And Publish Package +run-name: ${{ github.actor }} builds and publishes package to PyPI +on: + push: + branches: + - latest + +jobs: + publish: + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Cache uv binary + uses: actions/cache@v4 + with: + path: ${{ github.workspace }}/uv + key: uv-${{ runner.os }} + restore-keys: uv-${{ runner.os }} + + - name: Cache uv dependencies + uses: actions/cache@v4 + with: + path: ${{ github.workspace }}/.cache/uv + key: uv-${{ runner.os }} + restore-keys: uv-${{ runner.os }} + + - name: Cache pre-commit + uses: actions/cache@v4 + with: + path: ~/.cache/pre-commit + key: pre-commit-cache-${{ runner.os }}-${{ hashFiles('.pre-commit-config.yaml') }} + restore-keys: pre-commit-cache-${{ runner.os }}- + + - name: Install uv + uses: astral-sh/setup-uv@v5 + with: + version: "0.7.8" + enable-cache: true + cache-local-path: ${{ github.workspace }}/.cache/uv + tool-dir: ${{ github.workspace }}/.cache/uv + tool-bin-dir: ${{ github.workspace }}/.cache/uv + cache-dependency-glob: "" + + - name: Set up Python + run: uv python install + + - name: Install the project + run: uv sync --all-extras --no-install-project --cache-dir ${{ github.workspace }}/.cache/uv + + - name: Linter & Formatter + run: uv run pre-commit run --all-files + + - name: Build Package + run: uv build --cache-dir ${{ github.workspace }}/.cache/uv + + - name: Publish to Gitea PyPI + run: | + uv publish \ + --index twitchclient \ + --token ${{ secrets.CI_TOKEN }} diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d48f6de --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +# Python-generated files +__pycache__/ +*.py[oc] +build/ +dist/ +wheels/ +*.egg-info + +# Virtual environments +.venv + +# Ruff +.ruff_cache + +# uv +uv.lock diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..a42ca3c --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,34 @@ +repos: + - repo: https://github.com/crate-ci/typos + rev: v1.36.3 + hooks: + - id: typos + + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.13.2 + hooks: + - id: ruff + args: [ --fix ] + - id: ruff-format + + - repo: https://github.com/RobertCraigie/pyright-python + rev: v1.1.405 + hooks: + - id: pyright + + - repo: https://github.com/pre-commit/pre-commit-hooks + rev: v6.0.0 + hooks: + - id: trailing-whitespace + - id: check-docstring-first + - id: check-added-large-files + - id: check-yaml + - id: debug-statements + - id: check-merge-conflict + - id: double-quote-string-fixer + - id: end-of-file-fixer + + - repo: meta + hooks: + - id: check-hooks-apply + - id: check-useless-excludes diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..3543bd4 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,100 @@ +[project] +name = "twitchclient" +version = "0.1.0" +description = "Client for Twitch API" +readme = "README.md" +authors = [ + { name = "Miwory", email = "miwory.uwu@gmail.com" } +] +requires-python = ">=3.13" +dependencies = [ + "aiohttpx>=1.3,<=2.0", + "pydantic>=2.12,<=2.13", +] + +[build-system] +requires = ["uv_build>=0.9.2,<0.10.0"] +build-backend = "uv_build" + +[project.optional-dependencies] +dev = [ + "ruff==0.14.2", + "pyright==1.1.406", + "poethepoet==0.37.0", + "pre-commit==4.3.0", +] + +[[tool.uv.index]] +name = "Miwory" +url = "https://git.miwory.dev/api/packages/Miwory/pypi/simple" + +[[tool.uv.index]] +name = "twitchclient" +url = "https://git.miwory.dev/api/packages/Miwory/pypi/simple" +publish-url = "https://git.miwory.dev/api/packages/Miwory/pypi" +explicit = true + +[tool.poe.tasks] +_git = "git add ." +_lint = "pre-commit run --all-files" + +lint = ["_git", "_lint"] +check = "uv pip ls --outdated" + +[tool.pyright] +venvPath = "." +venv = ".venv" +strictListInference = true +strictDictionaryInference = true +strictSetInference = true +deprecateTypingAliases = true +typeCheckingMode = "strict" +pythonPlatform = "All" + +[tool.ruff] +target-version = "py313" +line-length = 79 +fix = true + +[tool.ruff.lint] +preview = true +select = [ + "E", + "W", + "F", + "UP", + "A", + "B", + "C4", + "SIM", + "I", + "S", + "G", + "FAST", + "ASYNC", + "BLE", + "INT", + "ISC", + "ICN", + "PYI", + "INP", + "RSE", + "PIE", + "SLOT", + "TID", + "LOG", + "FBT", + "DTZ", + "EM", + "PERF", + "RUF", +] +ignore = ["RUF029", "S101", "S104"] + +[tool.ruff.format] +quote-style = "single" +indent-style = "space" +docstring-code-format = true + +[tool.uv.sources] +aiohttpx = { index = "Miwory" } diff --git a/src/twitchclient/__init__.py b/src/twitchclient/__init__.py new file mode 100644 index 0000000..b794fd4 --- /dev/null +++ b/src/twitchclient/__init__.py @@ -0,0 +1 @@ +__version__ = '0.1.0' diff --git a/src/twitchclient/api.py b/src/twitchclient/api.py new file mode 100644 index 0000000..f5f7260 --- /dev/null +++ b/src/twitchclient/api.py @@ -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') diff --git a/src/twitchclient/py.typed b/src/twitchclient/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/twitchclient/schema.py b/src/twitchclient/schema.py new file mode 100644 index 0000000..e7d4374 --- /dev/null +++ b/src/twitchclient/schema.py @@ -0,0 +1,477 @@ +from datetime import datetime +from typing import Any, Literal, TypedDict + +from pydantic import BaseModel, ConfigDict, Field + + +class Error(Exception): + status_code: int + error: str + + def __init__(self, status_code: int, error: str) -> None: + self.status_code = status_code + self.error = error + super().__init__(f'{status_code}: {error}') + + +class Pagination(BaseModel): + model_config = ConfigDict(extra='forbid') + + cursor: str + + +class StartCommercialData(BaseModel): + model_config = ConfigDict(extra='forbid') + + length: int + message: str + retry_after: int + + +class StartCommercial(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[StartCommercialData] + + +class AdScheduleData(BaseModel): + model_config = ConfigDict(extra='forbid') + + next_ad_at: datetime | None + last_ad_at: datetime | None + duration: int + preroll_free_time: int + snooze_count: int + snooze_refresh_at: datetime + + +class AdSchedule(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[AdScheduleData] + + +class SnoozeNextAdData(BaseModel): + model_config = ConfigDict(extra='forbid') + + snooze_count: int + snooze_refresh_at: datetime + next_ad_at: datetime + + +class SnoozeNextAd(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[SnoozeNextAdData] + + +class DateRange(BaseModel): + model_config = ConfigDict(extra='forbid') + + started_at: datetime + ended_at: datetime + + +class ExtensionAnalyticsData(BaseModel): + model_config = ConfigDict(extra='forbid') + + extension_id: str + URL: str + type: str + date_range: DateRange + + +class ExtensionAnalytics(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[ExtensionAnalyticsData] + pagination: Pagination | dict[Any, Any] | None = None + + +class GameAnalyticsData(BaseModel): + model_config = ConfigDict(extra='forbid') + + game_id: int + URL: str + type: str + date_range: DateRange + + +class GameAnalytics(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[GameAnalyticsData] + pagination: Pagination | dict[Any, Any] | None = None + + +class BitsLeaderboardData(BaseModel): + model_config = ConfigDict(extra='forbid') + + user_id: int + user_login: str + user_name: str + rank: int + score: int + + +class BitsLeaderboard(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[BitsLeaderboardData] + date_range: DateRange + total: int + + +class CheermotesImageAnimated(BaseModel): + field_1: str = Field(..., alias='1') + field_1_5: str = Field(..., alias='1.5') + field_2: str = Field(..., alias='2') + field_3: str = Field(..., alias='3') + field_4: str = Field(..., alias='4') + + +class CheermotesImageStatic(BaseModel): + field_1: str = Field(..., alias='1') + field_1_5: str = Field(..., alias='1.5') + field_2: str = Field(..., alias='2') + field_3: str = Field(..., alias='3') + field_4: str = Field(..., alias='4') + + +class CheermotesImage(BaseModel): + animated: CheermotesImageAnimated + static: CheermotesImageStatic + + +class CheermotesImages(BaseModel): + model_config = ConfigDict(extra='forbid') + + light: CheermotesImage + dark: CheermotesImage + + +class CheermotesTier(BaseModel): + model_config = ConfigDict(extra='forbid') + + min_bits: int + id: Literal['1', '100', '500', '1000', '5000', '10000', '100000'] + color: str + can_cheer: bool + show_in_bits_card: bool + images: CheermotesImages + + +class CheermotesData(BaseModel): + model_config = ConfigDict(extra='forbid') + + prefix: str + tiers: list[CheermotesTier] + type: Literal[ + 'global_first_party', + 'global_third_party', + 'channel_custom', + 'display_only', + 'sponsored', + ] + order: int + last_updated: datetime + is_charitable: bool + + +class Cheermotes(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[CheermotesData] + + +class ExtensionProductCost(BaseModel): + model_config = ConfigDict(extra='forbid') + + amount: int + type: Literal['bits'] + + +class ExtensionProductData(BaseModel): + model_config = ConfigDict(extra='forbid') + + domain: str + sku: str + cost: ExtensionProductCost + inDevelopment: bool + displayName: str + expiration: str + broadcast: bool + + +class ExtensionTransactionsData(BaseModel): + model_config = ConfigDict(extra='forbid') + + id: str + timestamp: datetime + broadcaster_id: int + broadcaster_login: str + broadcaster_name: str + user_id: int + user_login: str + user_name: str + product_type: Literal['BITS_IN_EXTENSION'] + product_data: ExtensionProductData + + +class ExtensionTransactions(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[ExtensionTransactionsData] + pagination: Pagination | dict[Any, Any] | None = None + + +class ContentClassificationLabel(TypedDict): + id: Literal[ + 'DebatedSocialIssuesAndPolitics', + 'DrugsIntoxication', + 'SexualThemes', + 'ViolentGraphic', + 'Gambling', + 'ProfanityVulgarity', + ] + is_enabled: bool + + +class ChannelInformation(BaseModel): + model_config = ConfigDict(extra='forbid') + + broadcaster_id: int + broadcaster_login: str + broadcaster_name: str + broadcaster_language: str + game_name: str + game_id: int | str + title: str + delay: int + tags: list[str] + content_classification_labels: list[str] + is_branded_content: bool + + +class ChannelsInformation(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[ChannelInformation] + + +class ChannelEditor(BaseModel): + model_config = ConfigDict(extra='forbid') + + user_id: int + user_name: str + created_at: datetime + + +class ChannelEditors(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[ChannelEditor] + + +class FollowedChannel(BaseModel): + model_config = ConfigDict(extra='forbid') + + broadcaster_id: int + broadcaster_login: str + broadcaster_name: str + followed_at: datetime + + +class FollowedChannels(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[FollowedChannel] + pagination: Pagination | dict[Any, Any] | None = None + total: int + + +class ChannelFollower(BaseModel): + model_config = ConfigDict(extra='forbid') + + user_id: int + user_login: str + user_name: str + followed_at: datetime + + +class ChannelFollowers(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[ChannelFollower] + pagination: Pagination | dict[Any, Any] | None = None + total: int + + +class CustomRewardImage(BaseModel): + model_config = ConfigDict(extra='forbid') + + url_1x: str + url_2x: str + url_4x: str + + +class MaxPerStreamSetting(BaseModel): + model_config = ConfigDict(extra='forbid') + + is_enabled: bool + max_per_stream: int + + +class MaxPerUserPerStreamSetting(BaseModel): + model_config = ConfigDict(extra='forbid') + + is_enabled: bool + max_per_user_per_stream: int + + +class GlobalCooldownSetting(BaseModel): + model_config = ConfigDict(extra='forbid') + + is_enabled: bool + global_cooldown_seconds: int + + +class CustomReward(BaseModel): + model_config = ConfigDict(extra='forbid') + + broadcaster_id: int + broadcaster_login: str + broadcaster_name: str + id: int + title: str + prompt: str + cost: int + is_paused: bool + is_in_stock: bool + background_color: str + is_enabled: bool + is_user_input_required: bool + should_redemptions_skip_request_queue: bool + redemptions_redeemed_current_stream: int | None + cooldown_expires_at: datetime | None + image: CustomRewardImage | None + default_image: CustomRewardImage + max_per_stream_setting: MaxPerStreamSetting + max_per_user_per_stream_setting: MaxPerUserPerStreamSetting + global_cooldown_setting: GlobalCooldownSetting + + +class CustomRewards(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[CustomReward] + + +class CustomRewardRedemptionReward(BaseModel): + model_config = ConfigDict(extra='forbid') + + id: str + title: str + prompt: str + cost: int + + +class CustomRewardRedemption(BaseModel): + model_config = ConfigDict(extra='forbid') + + id: int + broadcaster_id: int + broadcaster_login: str + broadcaster_name: str + user_id: int + user_login: str + user_name: str + user_input: str + status: Literal['CANCELED', 'FULFILLED', 'UNFULFILLED'] + redeemed_at: datetime + reward: CustomRewardRedemptionReward + + +class CustomRewardRedemptions(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[CustomRewardRedemption] + + +class CharityCampaignCurrentAmount(BaseModel): + model_config = ConfigDict(extra='forbid') + + amount: int + decimal_places: int + currency: str + + +class CharityCampaignTargetAmount(BaseModel): + model_config = ConfigDict(extra='forbid') + + value: int + decimal_places: int + currency: str + + +class CharityCampaignData(BaseModel): + model_config = ConfigDict(extra='forbid') + + id: str + broadcaster_id: int + broadcaster_login: str + broadcaster_name: str + charity_name: str + charity_description: str + charity_logo: str + charity_website: str + current_amount: CharityCampaignCurrentAmount + target_amount: CharityCampaignTargetAmount + + +class CharityCampaign(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[CharityCampaignData] + + +class CharityDonationAmount(BaseModel): + model_config = ConfigDict(extra='forbid') + + value: int + decimal_places: int + currency: str + + +class CharityDonation(BaseModel): + model_config = ConfigDict(extra='forbid') + + id: str + campaign_id: str + user_id: int + user_login: str + user_name: str + amount: CharityDonationAmount + + +class CharityDonations(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[CharityDonation] + pagination: Pagination | dict[Any, Any] | None = None + + +class ChattersData(BaseModel): + model_config = ConfigDict(extra='forbid') + + user_id: int + user_login: str + user_name: str + + +class Chatters(BaseModel): + model_config = ConfigDict(extra='forbid') + + data: list[ChattersData] + pagination: Pagination | dict[Any, Any] | None = None + total: int