Lint/formatting fixes

tap_spotify/__init__.py

@@ -0,0 +1 @@
+"""Tap for Spotify."""

tap_spotify/auth.py

@@ -1,12 +1,14 @@
 """Spotify Authentication."""
 
 from singer_sdk.authenticators import OAuthAuthenticator, SingletonMeta
+from typing_extensions import Self, override
 
 
 class SpotifyAuthenticator(OAuthAuthenticator, metaclass=SingletonMeta):
     """Authenticator class for Spotify."""
 
     @property
+    @override
     def oauth_request_body(self):
         return {
             "grant_type": "refresh_token",
@@ -16,7 +18,8 @@ def oauth_request_body(self):
         }
 
     @classmethod
-    def create_for_stream(cls, stream):
+    def create_for_stream(cls, stream) -> Self:
+        """Create authenticator instance for a stream."""
         return cls(
             stream=stream,
             auth_endpoint="https://accounts.spotify.com/api/token",

tap_spotify/client.py

@@ -1,11 +1,11 @@
 """REST client handling, including SpotifyStream base class."""
 
-from typing import Iterable, Optional
-from urllib.parse import ParseResult, parse_qsl
-
 from functools import cached_property
+from typing import Iterable
+from urllib.parse import parse_qsl
 
 from singer_sdk.streams import RESTStream
+from typing_extensions import override
 
 from tap_spotify.auth import SpotifyAuthenticator
 from tap_spotify.pagination import BodyLinkPaginator
@@ -19,17 +19,20 @@ class SpotifyStream(RESTStream):
     chunk_size = None
 
     @cached_property
+    @override
     def authenticator(self):
         return SpotifyAuthenticator.create_for_stream(self)
 
+    @override
     def get_new_paginator(self):
         return BodyLinkPaginator()
 
-    def get_url_params(self, context, next_page_token: Optional[ParseResult]):
+    @override
+    def get_url_params(self, context, next_page_token):
         params = super().get_url_params(context, next_page_token)
         return dict(parse_qsl(next_page_token.query)) if next_page_token else params
 
-    def chunk_records(self, records: Iterable[dict]):
+    def chunk_records(self, records: Iterable[dict]):  # noqa: D102
         if not self.chunk_size:
             return [records]
 

tap_spotify/pagination.py

@@ -1,7 +1,13 @@
+"""Pagination classes for tap-spotify."""
+
 from singer_sdk.pagination import BaseHATEOASPaginator
+from typing_extensions import override
 
 
 class BodyLinkPaginator(BaseHATEOASPaginator):
+    """Body `next` link paginator."""
+
+    @override
     def get_next_url(self, response):
         data: dict = response.json()
         return data.get("next")

tap_spotify/schemas/__init__.py

@@ -0,0 +1 @@
+"""Schema definitions for tap-spotify."""

tap_spotify/schemas/album.py

@@ -1,4 +1,4 @@
-"""Schema definitions for album objects"""
+"""Schema definitions for album objects."""
 
 from singer_sdk import typing as th
 

tap_spotify/schemas/artist.py

@@ -1,4 +1,4 @@
-"""Schema definitions for artist objects"""
+"""Schema definitions for artist objects."""
 
 from singer_sdk import typing as th
 

tap_spotify/schemas/audio_features.py

@@ -1,4 +1,4 @@
-"""Schema definitions for audio features objects"""
+"""Schema definitions for audio features objects."""
 
 from singer_sdk.typing import (
     IntegerType,
@@ -12,12 +12,6 @@
 
 
 class AudioFeaturesObject(CustomObject):
-    """
-    https://developer.spotify.com/documentation/web-api/reference/#/operations/get-audio-features
-
-    https://developer.spotify.com/documentation/web-api/reference/#/operations/get-several-audio-features
-    """
-
     properties = PropertiesList(
         Property("acousticness", NumberType),
         Property("analysis_url", StringType),

tap_spotify/schemas/external.py

@@ -1,4 +1,4 @@
-"""Schema definitions for external objects"""
+"""Schema definitions for external objects."""
 
 from singer_sdk import typing as th
 

tap_spotify/schemas/followers.py

@@ -1,4 +1,4 @@
-"""Schema definitions for followers objects"""
+"""Schema definitions for followers objects."""
 
 from singer_sdk import typing as th
 

tap_spotify/schemas/image.py

@@ -1,4 +1,4 @@
-"""Schema definitions for image objects"""
+"""Schema definitions for image objects."""
 
 from singer_sdk import typing as th
 

tap_spotify/schemas/restriction.py

@@ -1,4 +1,4 @@
-"""Schema definitions for restriction objects"""
+"""Schema definitions for restriction objects."""
 
 from singer_sdk import typing as th
 

tap_spotify/schemas/track.py

@@ -1,4 +1,4 @@
-"""Schema definitions for track objects"""
+"""Schema definitions for track objects."""
 
 from singer_sdk import typing as th
 
@@ -23,7 +23,7 @@ class TrackObject(CustomObject):
         th.Property("id", th.StringType),
         th.Property("is_local", th.BooleanType),
         th.Property("is_playable", th.BooleanType),
-        # th.Property("linked_from", TrackObject),
+        # th.Property("linked_from", TrackObject),  # noqa: ERA001
         th.Property("name", th.StringType),
         th.Property("popularity", th.IntegerType),
         th.Property("preview_url", th.StringType),

tap_spotify/schemas/utils/__init__.py

@@ -0,0 +1 @@
+"""Schema utils for tap-spotify."""

tap_spotify/schemas/utils/custom_object.py

@@ -1,23 +1,31 @@
-"""Base custom object defintion"""
+"""Base custom object defintion."""
+
+from __future__ import annotations
 
 from singer_sdk import typing as th
-from singer_sdk.helpers._classproperty import classproperty
+from typing_extensions import Self, override
+
+# ruff: noqa: N805
 
 
 class CustomObject(th.JSONTypeHelper):
+    """Custom object."""
+
     properties: th.PropertiesList
 
-    @classproperty
+    @th.DefaultInstanceProperty
+    @override
     def type_dict(cls):
         return cls.properties.to_dict()
 
-    @classproperty
-    def schema(cls):
+    @th.DefaultInstanceProperty
+    def schema(cls):  # noqa: D102
         return cls.type_dict
 
     @classmethod
-    def extend_with(cls, *extras: "CustomObject"):
+    def extend_with(cls, *extras: type[Self]) -> type[Self]:
+        """Extend a custom object schema with other custom object types."""
         for e in extras:
-            for _, p in e.properties.items():
+            for _, p in e.properties.items():  # noqa: PERF102
                 cls.properties.append(p)
         return cls

tap_spotify/schemas/utils/rank.py

@@ -1,4 +1,4 @@
-"""Schema definition for rank schema wrapper"""
+"""Schema definition for rank schema wrapper."""
 
 from singer_sdk import typing as th
 

tap_spotify/schemas/utils/synced_at.py

@@ -1,4 +1,4 @@
-"""Schema definition for synced at schema wrapper"""
+"""Schema definition for synced at schema wrapper."""
 
 from singer_sdk import typing as th
 

tap_spotify/streams.py

@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Iterable
 
 from singer_sdk.streams.rest import RESTStream
+from typing_extensions import override
 
 from tap_spotify.client import SpotifyStream
 from tap_spotify.schemas.artist import ArtistObject
@@ -16,25 +17,21 @@
 
 
 class _RankStream(RESTStream):
-    """Define a rank stream."""
-
     rank = 1
 
+    @override
     def post_process(self, row, context):
-        """Apply rank integer to stream"""
         row = super().post_process(row, context)
         row["rank"] = self.rank
         self.rank += 1
         return row
 
 
 class _SyncedAtStream(RESTStream):
-    """Define a synced at stream."""
-
-    synced_at = datetime.utcnow()
+    synced_at = datetime.now(tz=timezone.utc)
 
+    @override
     def post_process(self, row, context):
-        """Apply synced at datetime to stream"""
         row = super().post_process(row, context)
         row["synced_at"] = self.synced_at
         return row
@@ -49,18 +46,24 @@ class _AudioFeaturesStream(SpotifyStream):
     schema = AudioFeaturesObject.schema
     max_tracks = 100
 
-    def __init__(self, tracks_stream: _TracksStream, track_records: Iterable[dict]):
-        super().__init__(tracks_stream._tap)
-
-        total_tracks = len(track_records)
-
-        if total_tracks > self.max_tracks:
-            msg = f"Cannot get audio features for more than {self.max_tracks} tracks at a time: {total_tracks} requested"
+    def __init__(
+        self,
+        tracks_stream: _TracksStream,
+        track_records: Iterable[dict],
+    ) -> None:
+        super().__init__(tracks_stream._tap)  # noqa: SLF001
+
+        if total_tracks := len(track_records) > self.max_tracks:
+            msg = (
+                f"Cannot get audio features for more than {self.max_tracks} tracks at a"
+                f" time: {total_tracks} requested"
+            )
             raise ValueError(msg)
 
         self._track_records = track_records
 
-    def get_url_params(self, *args, **kwargs):
+    @override
+    def get_url_params(self, context, next_page_token):
         return {"ids": ",".join([track["id"] for track in self._track_records])}
 
 
@@ -136,14 +139,14 @@ class UserTopTracksShortTermStream(
     """Define user top tracks short-term stream."""
 
     name = "user_top_tracks_st_stream"
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
 
 class UserTopTracksMediumTermStream(_UserTopTracksStream):
     """Define user top tracks medium-term stream."""
 
     name = "user_top_tracks_mt_stream"
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
 
 class UserTopTracksLongTermStream(
@@ -153,7 +156,7 @@ class UserTopTracksLongTermStream(
     """Define user top tracks long-term stream."""
 
     name = "user_top_tracks_lt_stream"
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
 
 class UserTopArtistsShortTermStream(
@@ -163,14 +166,14 @@ class UserTopArtistsShortTermStream(
     """Define user top artists short-term stream."""
 
     name = "user_top_artists_st_stream"
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
 
 class UserTopArtistsMediumTermStream(_UserTopArtistsStream):
     """Define user top artists medium-term stream."""
 
     name = "user_top_artists_mt_stream"
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
 
 class UserTopArtistsLongTermStream(
@@ -180,15 +183,15 @@ class UserTopArtistsLongTermStream(
     """Define user top artists long-term stream."""
 
     name = "user_top_artists_lt_stream"
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
 
 class _PlaylistTracksStream(_RankStream, _SyncedAtStream, _TracksStream):
     """Define playlist tracks stream."""
 
     records_jsonpath = "$.tracks.items[*].track"
     schema = TrackObject.extend_with(Rank, SyncedAt, AudioFeaturesObject).schema
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
     def parse_response(self, response):
         for track in super().parse_response(response):
@@ -201,31 +204,31 @@ class GlobalTopTracksDailyStream(_PlaylistTracksStream):
 
     name = "global_top_tracks_daily_stream"
     path = "/playlists/37i9dQZEVXbMDoHDwVN2tF"
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
 
 class GlobalTopTracksWeeklyStream(_PlaylistTracksStream):
     """Define global top tracks weekly stream."""
 
     name = "global_top_tracks_weekly_stream"
     path = "/playlists/37i9dQZEVXbNG2KDcFcKOF"
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
 
 class GlobalViralTracksDailyStream(_PlaylistTracksStream):
     """Define global viral tracks daily stream."""
 
     name = "global_viral_tracks_daily_stream"
     path = "/playlists/37i9dQZEVXbLiRSasKsNU9"
-    primary_keys = ["rank", "synced_at"]
+    primary_keys = ("rank", "synced_at")
 
 
 class UserSavedTracksStream(_SyncedAtStream, SpotifyStream):
     """Define user saved tracks stream."""
 
     name = "user_saved_tracks_stream"
     path = "/me/tracks"
-    primary_keys = ["id", "synced_at"]
+    primary_keys = ("id", "synced_at")
     limit = 50
     schema = TrackObject.extend_with(SyncedAt).schema
     records_jsonpath = "$.items[*].track"

tap_spotify/tap.py

@@ -1,8 +1,8 @@
 """Spotify tap class."""
 
-
 from singer_sdk import Tap
 from singer_sdk import typing as th
+from typing_extensions import override
 
 from tap_spotify import streams
 
@@ -48,6 +48,7 @@ class TapSpotify(Tap):
         ),
     ).to_dict()
 
+    @override
     def discover_streams(self):
         return [stream_class(tap=self) for stream_class in STREAM_TYPES]