From 727af5899a42baea00816c9c8edc36f950e6181f Mon Sep 17 00:00:00 2001 From: Miwory Date: Sun, 8 Mar 2026 06:41:33 +0300 Subject: [PATCH] =?UTF-8?q?=D0=9F=D0=B5=D1=80=D0=B2=D1=8B=D0=B9=20=D1=80?= =?UTF-8?q?=D0=B5=D0=BB=D0=B8=D0=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitea/workflows/dev.yaml | 58 +++++ .gitea/workflows/latest.yaml | 64 +++++ .gitignore | 16 ++ .pre-commit-config.yaml | 34 +++ README.md | 100 ++++++++ pyproject.toml | 99 ++++++++ src/oxidespotify/__init__.py | 0 src/oxidespotify/api.py | 439 +++++++++++++++++++++++++++++++++++ src/oxidespotify/auth.py | 70 ++++++ src/oxidespotify/py.typed | 0 src/oxidespotify/schema.py | 353 ++++++++++++++++++++++++++++ 11 files changed, 1233 insertions(+) create mode 100644 .gitea/workflows/dev.yaml create mode 100644 .gitea/workflows/latest.yaml create mode 100644 .gitignore create mode 100644 .pre-commit-config.yaml create mode 100644 README.md create mode 100644 pyproject.toml create mode 100644 src/oxidespotify/__init__.py create mode 100644 src/oxidespotify/api.py create mode 100644 src/oxidespotify/auth.py create mode 100644 src/oxidespotify/py.typed create mode 100644 src/oxidespotify/schema.py diff --git a/.gitea/workflows/dev.yaml b/.gitea/workflows/dev.yaml new file mode 100644 index 0000000..4f2f8cf --- /dev/null +++ b/.gitea/workflows/dev.yaml @@ -0,0 +1,58 @@ +name: Verify Dev Build +run-name: ${{ github.actor }} verifying dev build +on: + push: + branches: + - dev + +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.10.5" + 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 diff --git a/.gitea/workflows/latest.yaml b/.gitea/workflows/latest.yaml new file mode 100644 index 0000000..866b646 --- /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.10.5" + 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 OxideSpotify \ + --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..5eb171f --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,34 @@ +repos: + - repo: https://github.com/crate-ci/typos + rev: v1.42.0 + hooks: + - id: typos + + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.15.0 + hooks: + - id: ruff + args: [ --fix ] + - id: ruff-format + + - repo: https://github.com/RobertCraigie/pyright-python + rev: v1.1.408 + 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..2baebc1 --- /dev/null +++ b/README.md @@ -0,0 +1,100 @@ +# 🟢 OxideSpotify + +**OxideSpotify** is a high-performance Spotify Web API client built on the [OxideHTTP](https://git.miwory.dev/OxideHTTP/OxideHTTP) core. It combines the efficiency of Rust-backed networking with a strictly typed Pythonic interface, designed to handle Spotify's unique requirements like automatic retry logic and complex search filtering. + +--- + +## 🔥 Why OxideSpotify? + +The Spotify API requires precision in header management and often returns `504 Gateway Timeout` or `204 No Content` for player operations. OxideSpotify manages these edge cases automatically: + +* **⚡ Oxide Core:** Powered by `pyreqwest` (Rust) for minimal overhead in HTTP calls. +* **🔄 Intelligent Retries:** Automatically re-processes requests on `504` errors to ensure stability in unstable network conditions. +* **🛡️ Rate Limit Resiliency:** Integrates with Redis via OxideHTTP to stay within Spotify’s variable rate limits. +* **🏗️ Advanced Overloads:** Provides precise type hinting for the `/search` endpoint—your IDE knows exactly which model is returned based on the `search_type` parameter. + +--- + +## 📦 Installation + +Configure your private registry in your `uv` environment, then run: + +```bash +uv add oxidespotify + +``` + +--- + +## 🛠 Quick Start + +### Playback Management + +Spotify's player endpoints frequently return `204 No Content`. OxideSpotify handles this transition gracefully. + +```python +import asyncio +from oxidespotify.api import SpotifyAPIClient + +async def main(): + async with SpotifyAPIClient( + redis_url="redis://localhost:6379" + ) as spotify: + # Get current playback state + state = await spotify.get_playback_state( + access_token="your_token", + additional_types=['track'] + ) + + if state: + print(f"Now Playing: {state.item.name} by {state.item.artists[0].name}") + + # Add to queue (Handles 204 successfully) + await spotify.add_item_to_playback_queue( + access_token="your_token", + uri="spotify:track:4cOdK2wGvS9II9Ja7pveUv", + device_id="your_device_id" + ) + +asyncio.run(main()) + +``` + +### Type-Safe Searching + +Thanks to Python `@overload`, your IDE provides full autocomplete for specific search results. + +```python +# The IDE knows 'results' is a SearchArtist model +results = await spotify.search_for_item( + access_token="token", + q="Daft Punk", + search_type="artist" +) + +for artist in results.artists.items: + print(artist.name, artist.genres) + +``` + +--- + +## ⚙️ Advanced: Configuration + +Ensure your `pyproject.toml` is configured to find the Oxide ecosystem in your registry: + +```toml +[[tool.uv.index]] +name = "OxideHTTP" +url = "https://git.miwory.dev/api/packages/OxideHTTP/pypi/simple" + +``` + +### Implementation Coverage + +The client currently provides full Pydantic models and logic for: + +* **Player:** Playback state, queue management, and device control. +* **Search:** Comprehensive filtering for albums, artists, tracks, playlists, etc. +* **Users:** Profile retrieval and recently played history. +* **Tracks:** Detailed track metadata retrieval. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d243da4 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,99 @@ +[project] +name = "oxidespotify" +version = "0.1.0" +description = "Client for Spotify API" +readme = "README.md" +authors = [{ name = "Miwory", email = "miwory.uwu@gmail.com" }] +requires-python = ">=3.14" +dependencies = ["oxidehttp>=1.0.3,<=2.0.0", "pydantic>=2.12,<=2.13"] + +[build-system] +requires = ["uv_build>=0.9.2,<0.11.0"] +build-backend = "uv_build" + +[dependency-groups] +dev = [ + "poethepoet>=0.40.0", + "pre-commit>=4.5.1", +] + +[[tool.uv.index]] +name = "OxideHTTP" +url = "https://git.miwory.dev/api/packages/OxideHTTP/pypi/simple" + +[[tool.uv.index]] +name = "OxideSpotify" +url = "https://git.miwory.dev/api/packages/OxideHTTP/pypi/simple" +publish-url = "https://git.miwory.dev/api/packages/OxideHTTP/pypi" +explicit = true + +[tool.poe.tasks] +_git = "git add ." +_lint = "pre-commit run --all-files" +_ty = "uvx ty@latest check ." + +lint = ["_git", "_lint", "_ty"] +check = "uv pip ls --outdated" + +major = "uv version --bump major" +minor = "uv version --bump minor" +patch = "uv version --bump patch" + +[tool.pyright] +venvPath = "." +venv = ".venv" +strictListInference = true +strictDictionaryInference = true +strictSetInference = true +deprecateTypingAliases = true +typeCheckingMode = "strict" +pythonPlatform = "All" +stubPath = "typings" + +[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 = ["RUF002", "RUF029", "S101", "S104", "W505"] + +[tool.ruff.lint.pydoclint] +ignore-one-line-docstrings = true + +[tool.ruff.format] +quote-style = "single" +indent-style = "space" +docstring-code-format = true diff --git a/src/oxidespotify/__init__.py b/src/oxidespotify/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/oxidespotify/api.py b/src/oxidespotify/api.py new file mode 100644 index 0000000..a38d260 --- /dev/null +++ b/src/oxidespotify/api.py @@ -0,0 +1,439 @@ +from collections.abc import Coroutine +from typing import Literal, NoReturn, overload + +from oxidehttp.client import OxideHTTP +from oxidehttp.schema import CachedResponse, Response +from pydantic import BaseModel + +from . import schema as s + + +class SpotifyAPIClient(OxideHTTP): + def __init__( + self, + redis_url: str | None = None, + proxy_url: str | None = None, + ) -> None: + super().__init__( + base_url='https://api.spotify.com/v1/', + redis_url=redis_url, + proxy_url=proxy_url, + ) + + def _auth(self, access_token: str) -> dict[str, str]: + return {'Authorization': f'Bearer {access_token}'} + + async def _process_request[T: Response | CachedResponse]( + self, func: Coroutine[None, None, T] + ) -> T: + req = await func + + if req.status_code == 504: + return await self._process_request(func) + + return req + + async def _process[T: BaseModel]( + self, req: Response | CachedResponse, schema: type[T] | None + ) -> T | None: + if req.status_code == 204: + return None + + if req.status_code >= 500: + raise s.InternalError(req.status_code, 'Internal server error') + + data = await req.json() + if req.status_code >= 400: + raise s.ClientError(req.status_code, data['error']['message']) + + if schema: + return schema.model_validate(data) + + async def get_album(self) -> NoReturn: + raise NotImplementedError + + async def get_album_tracks(self) -> NoReturn: + raise NotImplementedError + + async def get_users_saved_albums(self) -> NoReturn: + raise NotImplementedError + + async def get_artist(self) -> NoReturn: + raise NotImplementedError + + async def get_artists_albums(self) -> NoReturn: + raise NotImplementedError + + async def get_an_audiobook(self) -> NoReturn: + raise NotImplementedError + + async def get_audiobook_chapters(self) -> NoReturn: + raise NotImplementedError + + async def get_users_saved_audiobooks(self) -> NoReturn: + raise NotImplementedError + + async def get_a_chapter(self) -> NoReturn: + raise NotImplementedError + + async def get_episode(self) -> NoReturn: + raise NotImplementedError + + async def get_users_saved_episodes(self) -> NoReturn: + raise NotImplementedError + + async def save_item_to_library(self) -> NoReturn: + raise NotImplementedError + + async def remove_item_from_library(self) -> NoReturn: + raise NotImplementedError + + async def check_users_saved_items(self) -> NoReturn: + raise NotImplementedError + + async def get_playback_state( + self, + access_token: str, + *, + market: str | None = None, + additional_types: list[Literal['track', 'episode']] | None = None, + cache_ttl: int | None = None, + ) -> s.PlaybackState | None: + req = self.get( + '/me/player', + cache_ttl=cache_ttl, + params=self.clean_dict( + { + 'market': market, + 'additional_types': ','.join(additional_types) + if additional_types + else None, + } + ), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + return await self._process(req, s.PlaybackState) + + async def transfer_playback(self) -> NoReturn: + raise NotImplementedError + + async def get_available_devices(self) -> NoReturn: + raise NotImplementedError + + async def start_resume_playback(self) -> NoReturn: + raise NotImplementedError + + async def pause_playback(self) -> NoReturn: + raise NotImplementedError + + async def skip_to_next(self) -> NoReturn: + raise NotImplementedError + + async def skip_to_previous(self) -> NoReturn: + raise NotImplementedError + + async def seek_to_position(self) -> NoReturn: + raise NotImplementedError + + async def set_repeat_mode(self) -> NoReturn: + raise NotImplementedError + + async def set_playback_volume(self) -> NoReturn: + raise NotImplementedError + + async def toggle_playback_shuffle(self) -> NoReturn: + raise NotImplementedError + + async def get_recently_played_tracks( + self, + access_token: str, + *, + limit: int = 20, + after: int | None = None, + before: int | None = None, + cache_ttl: int | None = None, + ) -> s.RecentlyPlayed: + req = self.get( + '/me/player/recently-played', + cache_ttl=cache_ttl, + params=self.clean_dict( + {'limit': limit, 'after': after, 'before': before} + ), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + res = await self._process(req, s.RecentlyPlayed) + + if res is None: + raise s.InternalError(status_code=0, error="Can't be here") + + return res + + async def get_the_users_queue(self) -> NoReturn: + raise NotImplementedError + + async def add_item_to_playback_queue( + self, access_token: str, uri: str, device_id: str + ) -> None: + req = self.post( + '/me/player/queue', + params={'uri': uri, 'device_id': device_id}, + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + res = await self._process(req, None) + + return res + + async def get_playlist(self) -> NoReturn: + raise NotImplementedError + + async def change_playlist_details(self) -> NoReturn: + raise NotImplementedError + + async def get_playlist_items(self) -> NoReturn: + raise NotImplementedError + + async def update_playlist_items(self) -> NoReturn: + raise NotImplementedError + + async def add_items_to_playlist(self) -> NoReturn: + raise NotImplementedError + + async def remove_playlist_items(self) -> NoReturn: + raise NotImplementedError + + async def get_current_users_playlists(self) -> NoReturn: + raise NotImplementedError + + async def create_playlist(self) -> NoReturn: + raise NotImplementedError + + async def get_playlist_cover_image(self) -> NoReturn: + raise NotImplementedError + + async def add_custom_playlist_cover_image(self) -> NoReturn: + raise NotImplementedError + + @overload + async def search_for_item( + self, + access_token: str, + q: str, + search_type: Literal['album'], + *, + market: str | None = None, + limit: int = 5, + offset: int = 0, + include_external: Literal['audio'] | None = None, + cache_ttl: int | None = None, + ) -> s.SearchAlbum: ... + + @overload + async def search_for_item( + self, + access_token: str, + q: str, + search_type: Literal['artist'], + *, + market: str | None = None, + limit: int = 5, + offset: int = 0, + include_external: Literal['audio'] | None = None, + cache_ttl: int | None = None, + ) -> s.SearchArtist: ... + + @overload + async def search_for_item( + self, + access_token: str, + q: str, + search_type: Literal['playlist'], + *, + market: str | None = None, + limit: int = 5, + offset: int = 0, + include_external: Literal['audio'] | None = None, + cache_ttl: int | None = None, + ) -> s.SearchPlaylist: ... + + @overload + async def search_for_item( + self, + access_token: str, + q: str, + search_type: Literal['track'], + *, + market: str | None = None, + limit: int = 5, + offset: int = 0, + include_external: Literal['audio'] | None = None, + cache_ttl: int | None = None, + ) -> s.SearchTrack: ... + + @overload + async def search_for_item( + self, + access_token: str, + q: str, + search_type: Literal['show'], + *, + market: str | None = None, + limit: int = 5, + offset: int = 0, + include_external: Literal['audio'] | None = None, + cache_ttl: int | None = None, + ) -> s.SearchShow: ... + + @overload + async def search_for_item( + self, + access_token: str, + q: str, + search_type: Literal['episode'], + *, + market: str | None = None, + limit: int = 5, + offset: int = 0, + include_external: Literal['audio'] | None = None, + cache_ttl: int | None = None, + ) -> s.SearchEpisode: ... + + @overload + async def search_for_item( + self, + access_token: str, + q: str, + search_type: Literal['audiobook'], + *, + market: str | None = None, + limit: int = 5, + offset: int = 0, + include_external: Literal['audio'] | None = None, + cache_ttl: int | None = None, + ) -> s.SearchAudiobook: ... + + async def search_for_item( + self, + access_token: str, + q: str, + search_type: Literal[ + 'album', + 'artist', + 'playlist', + 'track', + 'show', + 'episode', + 'audiobook', + ], + *, + market: str | None = None, + limit: int = 5, + offset: int = 0, + include_external: Literal['audio'] | None = None, + cache_ttl: int | None = None, + ) -> ( + s.SearchAlbum + | s.SearchArtist + | s.SearchTrack + | s.SearchPlaylist + | s.SearchShow + | s.SearchEpisode + | s.SearchAudiobook + ): + type_mapping = { + 'album': s.SearchAlbum, + 'artist': s.SearchArtist, + 'track': s.SearchTrack, + 'playlist': s.SearchPlaylist, + 'show': s.SearchShow, + 'episode': s.SearchEpisode, + 'audiobook': s.SearchAudiobook, + } + target_model = type_mapping[search_type] + + req = self.get( + '/search', + cache_ttl=cache_ttl, + params=self.clean_dict( + { + 'q': q, + 'type': search_type, + 'market': market, + 'limit': limit, + 'offset': offset, + 'include_external': include_external, + } + ), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + res = await self._process(req, target_model) + + if res is None: + raise s.InternalError(status_code=0, error="Can't be here") + + return res + + async def get_show(self) -> NoReturn: + raise NotImplementedError + + async def get_show_episodes(self) -> NoReturn: + raise NotImplementedError + + async def get_users_saved_shows(self) -> NoReturn: + raise NotImplementedError + + async def get_track( + self, + access_token: str, + track_id: str, + *, + market: str | None = None, + cache_ttl: int | None = None, + ) -> s.Track: + req = self.get( + f'/tracks/{track_id}', + cache_ttl=cache_ttl, + params=self.clean_dict({'market': market}), + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + res = await self._process(req, s.Track) + + if res is None: + raise s.InternalError(status_code=0, error="Can't be here") + + return res + + async def get_users_saved_tracks(self) -> NoReturn: + raise NotImplementedError + + async def get_current_users_profile( + self, access_token: str, *, cache_ttl: int | None = None + ) -> s.UserProfile: + req = self.get( + '/me', + cache_ttl=cache_ttl, + headers=self._auth(access_token), + ) + + req = await self._process_request(req) + res = await self._process(req, s.UserProfile) + + if res is None: + raise s.InternalError(status_code=0, error="Can' be here") + + return res + + async def get_users_top_items(self) -> NoReturn: + raise NotImplementedError + + async def get_followed_artists(self) -> NoReturn: + raise NotImplementedError diff --git a/src/oxidespotify/auth.py b/src/oxidespotify/auth.py new file mode 100644 index 0000000..745b6df --- /dev/null +++ b/src/oxidespotify/auth.py @@ -0,0 +1,70 @@ +from base64 import b64encode + +from oxidehttp.client import OxideHTTP +from oxidehttp.schema import CachedResponse, Response +from pydantic import BaseModel + +from . import schema as s + + +class SpotifyAuthClient(OxideHTTP): + def __init__( + self, + client_id: str, + client_secret: str, + redirect_uri: str, + redis_url: str | None = None, + proxy_url: str | None = None, + ) -> None: + self.__client_id = client_id + self.__client_secret = client_secret + self.__redirect_uri = redirect_uri + + credentials = f'{self.__client_id}:{self.__client_secret}' + encoded_credentials = b64encode(credentials.encode('utf-8')).decode() + + super().__init__( + base_url='https://accounts.spotify.com/api/', + headers={ + 'Content-Type': 'application/x-www-form-urlencoded', + 'Authorization': f'Basic {encoded_credentials}', + }, + redis_url=redis_url, + proxy_url=proxy_url, + ) + + async def _process[T: BaseModel]( + self, req: Response | CachedResponse, schema: type[T] + ) -> T: + if req.status_code >= 500: + raise s.InternalError(req.status_code, 'Internal server error') + + data = await req.json() + if req.status_code >= 400: + raise s.ClientError(req.status_code, data['error']['message']) + + return schema.model_validate(data) + + async def user_access_token(self, code: str) -> s.UserAccessToken: + req = await self.post( + '/token', + json={ + 'grant_type': 'authorization_code', + 'code': code, + 'redirect_uri': self.__redirect_uri, + }, + ) + + return await self._process(req, s.UserAccessToken) + + async def refresh_access_token(self, refresh_token: str) -> s.Token: + req = await self.post( + '/token', + json={ + 'grant_type': 'refresh_token', + 'refresh_token': refresh_token, + 'client_id': self.__client_id, + }, + ) + + return await self._process(req, s.Token) diff --git a/src/oxidespotify/py.typed b/src/oxidespotify/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/oxidespotify/schema.py b/src/oxidespotify/schema.py new file mode 100644 index 0000000..31e8145 --- /dev/null +++ b/src/oxidespotify/schema.py @@ -0,0 +1,353 @@ +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, HttpUrl, field_validator + + +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 ClientError(Error): + pass + + +class InternalError(Error): + pass + + +class Token(BaseModel): + token_type: Literal['Bearer'] + access_token: str + expires_in: int + scope: str + + +class UserAccessToken(Token): + refresh_token: str + + +class Device(BaseModel): + id: str | None + is_active: bool + is_private_session: bool + is_restricted: bool + name: str + type: str + volume_percent: int | None + supports_volume: bool + + +class Restriction(BaseModel): + reason: Literal['market', 'product', 'explicit'] + + +class Image(BaseModel): + url: HttpUrl + height: int + width: int + + @field_validator('height', 'width', mode='before') + @classmethod + def convert_null_to_zero(cls, v: int | None) -> int: + if v is None: + return 0 + + return v + + +class ExternalUrls(BaseModel): + spotify: HttpUrl + + +class ExternalIDs(BaseModel): + isrc: str | None = None + ean: str | None = None + upc: str | None = None + + +class Artist(BaseModel): + external_urls: ExternalUrls + href: HttpUrl + id: str + name: str + type: Literal['artist'] + uri: str + + +class Cursor(BaseModel): + after: int + before: int + + +class Context(BaseModel): + type: Literal['artist', 'playlist', 'album', 'show', 'collection'] + href: HttpUrl + external_urls: ExternalUrls + uri: str + + +class Album(BaseModel): + album_type: Literal['album', 'single', 'compilation'] + total_tracks: int + external_urls: ExternalUrls + href: HttpUrl + id: str + images: list[Image] + name: str + release_date: datetime + release_date_precision: Literal['year', 'month', 'day'] + restrictions: Restriction | None = None + type: Literal['album'] + uri: str + artists: list[Artist] + + +class PlaylistOwner(BaseModel): + external_urls: ExternalUrls + href: HttpUrl + id: str + type: Literal['user'] + uri: str + display_name: str | None + + +class PlaylistTracks(BaseModel): + href: HttpUrl + total: int + + +class Playlist(BaseModel): + collaborative: bool + description: str + external_urls: ExternalUrls + href: HttpUrl + id: str + images: list[Image] + name: str + owner: PlaylistOwner + public: bool + snapshot_id: str + items: PlaylistTracks + type: Literal['playlist'] + uri: str + + +class Track(BaseModel): + album: Album + artists: list[Artist] + disc_number: int + duration_ms: int + explicit: bool + external_ids: ExternalIDs + external_urls: ExternalUrls + href: HttpUrl + id: str + is_playable: bool | None = None + restrictions: Restriction | None = None + name: str + track_number: int + type: Literal['track'] + uri: str + is_local: bool + + +class EpisodeResumePoint(BaseModel): + fully_played: bool + resume_position_ms: int + + +class Copyright(BaseModel): + text: str + type: Literal['C', 'P'] + + +class Show(BaseModel): + copyrights: list[Copyright] + description: str + html_description: str + explicit: bool + external_ids: ExternalIDs | None = None + href: HttpUrl + id: str + images: list[Image] + is_externally_hosted: bool + languages: list[str] + media_type: str + name: str + type: Literal['show'] + uri: str + total_episodes: int + + +class Episode(BaseModel): + description: str + html_description: str + duration_ms: int + explicit: bool + external_urls: ExternalUrls + href: HttpUrl + id: str + images: list[Image] + is_externally_hosted: bool + is_playable: bool + languages: list[str] + name: str + release_date: datetime + release_date_precision: Literal['year', 'month', 'day'] + resume_point: EpisodeResumePoint + type: Literal['episode'] + uri: str + restrictions: Restriction | None = None + show: Show | None = None + + +class AudioBookAuthor(BaseModel): + name: str + + +class AudioBookNarator(BaseModel): + name: str + + +class AudioBook(BaseModel): + authors: list[AudioBookAuthor] + copyrights: list[Copyright] + description: str + html_description: str + edition: str + explicit: bool + external_urls: ExternalUrls + href: str + id: str + images: list[Image] + languages: list[str] + media_type: str + name: str + narrators: list[AudioBookNarator] + type: Literal['audiobook'] + uri: str + total_chapters: int + + +class PlaybackStateActions(BaseModel): + interrupting_playback: bool | None = None + pausing: bool | None = None + resuming: bool | None = None + seeking: bool | None = None + skipping_next: bool | None = None + skipping_prev: bool | None = None + toggling_repeat_context: bool | None = None + toggling_shuffle: bool | None = None + toggling_repeat_track: bool | None = None + transferring_playback: bool | None = None + + +class PlaybackState(BaseModel): + device: Device + repeat_state: Literal['off', 'track', 'context'] + shuffle_state: bool + context: Context | None + timestamp: int + progress_ms: int + is_playing: bool + item: Track | Episode | None + currently_playing_type: Literal['track', 'episode', 'ad', 'unknown'] + # TODO: allows has 'disallows' inside? not sure about that + # actions: PlaybackStateActions + + +class PlayHistoryObject(BaseModel): + track: Track + played_at: datetime + context: Context | None = None + + +class RecentlyPlayed(BaseModel): + href: HttpUrl + limit: int + next: HttpUrl | None + cursors: Cursor + total: int | None = None + items: list[PlayHistoryObject] + + +class SearchData(BaseModel): + href: HttpUrl + limit: int + next: HttpUrl | None + offset: int + previous: int | None + total: int + + +class SearchAlbumData(SearchData): + items: list[Album] + + +class SearchAlbum(BaseModel): + albums: SearchAlbumData + + +class SearchArtistData(SearchData): + items: list[Artist] + + +class SearchArtist(BaseModel): + artists: SearchArtistData + + +class SearchPlaylistData(SearchData): + items: list[Playlist | None] + + +class SearchPlaylist(BaseModel): + playlists: SearchPlaylistData + + +class SearchTrackData(SearchData): + items: list[Track] + + +class SearchTrack(BaseModel): + tracks: SearchTrackData + + +class SearchShowData(SearchData): + items: list[Show] + + +class SearchShow(BaseModel): + shows: SearchShowData + + +class SearchEpisodeData(SearchData): + items: list[Episode] + + +class SearchEpisode(BaseModel): + episodes: SearchEpisodeData + + +class SearchAudiobookData(SearchData): + items: list[AudioBook] + + +class SearchAudiobook(BaseModel): + audiobooks: SearchAudiobookData + + +class UserProfile(BaseModel): + id: str + display_name: str + external_urls: ExternalUrls + href: HttpUrl + images: list[Image] + type: Literal['user'] + uri: str