Add playback state

pyproject.toml

@@ -28,6 +28,8 @@ packages = [
 python = "^3.11"
 aiohttp = ">=3.0.0"
 yarl = ">=1.6.0"
+mashumaro = "^3.11"
+orjson = "^3.9.10"
 
 [tool.poetry.group.dev.dependencies]
 aresponses = "2.1.6"

src/spotifyaio/__init__.py

@@ -0,0 +1,17 @@
+"""Asynchronous Python client for Spotify."""
+from .exceptions import (
+    SpotifyAuthenticationFailedError,
+    SpotifyConnectionError,
+    SpotifyError,
+)
+from .models import Device, PlaybackState
+from .spotify import SpotifyClient
+
+__all__ = [
+    "Device",
+    "SpotifyError",
+    "SpotifyConnectionError",
+    "SpotifyAuthenticationFailedError",
+    "SpotifyClient",
+    "PlaybackState",
+]

src/spotifyaio/exceptions.py

@@ -0,0 +1,13 @@
+"""Asynchronous Python client for Spotify."""
+
+
+class SpotifyError(Exception):
+    """Generic exception."""
+
+
+class SpotifyConnectionError(SpotifyError):
+    """Spotify connection exception."""
+
+
+class SpotifyAuthenticationFailedError(SpotifyError):
+    """Spotify authentication failed exception."""

src/spotifyaio/models.py

@@ -0,0 +1,25 @@
+"""Models for Spotify."""
+from dataclasses import dataclass, field
+
+from mashumaro import field_options
+from mashumaro.mixins.orjson import DataClassORJSONMixin
+
+
+@dataclass
+class Device(DataClassORJSONMixin):
+    """Device model."""
+
+    device_id: str = field(metadata=field_options(alias="id"))
+    is_active: bool
+    is_private_session: bool
+    name: str
+    supports_volume: bool
+    device_type: str = field(metadata=field_options(alias="type"))
+    volume_percent: int
+
+
+@dataclass
+class PlaybackState(DataClassORJSONMixin):
+    """Playback state model."""
+
+    device: Device

src/spotifyaio/spotify.py

@@ -0,0 +1,113 @@
+"""Spotify client for handling connections with Spotify."""
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from importlib import metadata
+from typing import Any, Awaitable, Callable, Self
+
+from aiohttp import ClientSession
+from aiohttp.hdrs import METH_GET
+from yarl import URL
+
+from spotifyaio.exceptions import SpotifyConnectionError, SpotifyError
+from spotifyaio.models import PlaybackState
+
+
+@dataclass
+class SpotifyClient:
+    """Main class for handling connections with Spotify."""
+
+    session: ClientSession | None = None
+    request_timeout: int = 10
+    api_host: str = "api.spotify.com"
+    _token: str | None = None
+    _close_session: bool = False
+    refresh_token_function: Callable[[], Awaitable[str]] | None = None
+
+    async def refresh_token(self) -> None:
+        """Refresh token with provided function."""
+        if self.refresh_token_function:
+            self._token = await self.refresh_token_function()
+
+    def authenticate(self, token: str) -> None:
+        """Authenticate the user with a token."""
+        self._token = token
+
+    async def _request(
+        self,
+        uri: str,
+        *,
+        data: dict[str, Any] | None = None,
+    ) -> str:
+        """Handle a request to Spotify."""
+        version = metadata.version(__package__)
+        url = URL.build(
+            scheme="https",
+            host=self.api_host,
+            port=443,
+        ).joinpath(uri)
+
+        await self.refresh_token()
+
+        headers = {
+            "User-Agent": f"AioSpotify/{version}",
+            "Accept": "application/json, text/plain, */*",
+            "Authorization": f"Bearer {self._token}",
+        }
+
+        if self.session is None:
+            self.session = ClientSession()
+            self._close_session = True
+
+        try:
+            async with asyncio.timeout(self.request_timeout):
+                response = await self.session.request(
+                    METH_GET,
+                    url,
+                    headers=headers,
+                    data=data,
+                )
+        except asyncio.TimeoutError as exception:
+            msg = "Timeout occurred while connecting to Spotify"
+            raise SpotifyConnectionError(msg) from exception
+
+        content_type = response.headers.get("Content-Type", "")
+
+        if "application/json" not in content_type:
+            text = await response.text()
+            msg = "Unexpected response from Spotify"
+            raise SpotifyError(
+                msg,
+                {"Content-Type": content_type, "response": text},
+            )
+
+        return await response.text()
+
+    async def get_playback(self) -> PlaybackState:
+        """Get devices."""
+        response = await self._request("v1/me/player")
+        return PlaybackState.from_json(response)
+
+    async def close(self) -> None:
+        """Close open client session."""
+        if self.session and self._close_session:
+            await self.session.close()
+
+    async def __aenter__(self) -> Self:
+        """Async enter.
+
+        Returns
+        -------
+            The SpotifyClient object.
+        """
+        return self
+
+    async def __aexit__(self, *_exc_info: object) -> None:
+        """Async exit.
+
+        Args:
+        ----
+            _exc_info: Exec type.
+        """
+        await self.close()

tests/__init__.py

@@ -0,0 +1,8 @@
+"""Asynchronous Python client for Spotify."""
+from pathlib import Path
+
+
+def load_fixture(filename: str) -> str:
+    """Load a fixture."""
+    path = Path(__package__) / "fixtures" / filename
+    return path.read_text(encoding="utf-8")

tests/__snapshots__/test_playback.ambr

@@ -0,0 +1,14 @@
+# serializer version: 1
+# name: test_get_devices
+  dict({
+    'device': dict({
+      'device_id': '21dac6b0e0a1f181870fdc9749b2656466557687',
+      'device_type': 'Computer',
+      'is_active': True,
+      'is_private_session': False,
+      'name': 'DESKTOP-BKC5SIK',
+      'supports_volume': True,
+      'volume_percent': 69,
+    }),
+  })
+# ---

tests/conftest.py

@@ -0,0 +1,12 @@
+"""Asynchronous Python client for Spotify."""
+import pytest
+
+from syrupy import SnapshotAssertion
+
+from .syrupy import SpotifySnapshotExtension
+
+
+@pytest.fixture(name="snapshot")
+def snapshot_assertion(snapshot: SnapshotAssertion) -> SnapshotAssertion:
+    """Return snapshot assertion fixture with the Spotify extension."""
+    return snapshot.use_extension(SpotifySnapshotExtension)

tests/const.py

@@ -0,0 +1,3 @@
+"""Constants for tests."""
+
+SPOTIFY_URL = "api.spotify.com"