Sync audio features for tracks

tap_spotify/schemas/audio_features.py

@@ -0,0 +1,40 @@
+"""Schema definitions for audio features objects"""
+
+from singer_sdk.typing import (
+    IntegerType,
+    NumberType,
+    PropertiesList,
+    Property,
+    StringType,
+)
+
+from tap_spotify.schemas.utils.custom_object import CustomObject
+
+
+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),
+        Property("danceability", NumberType),
+        Property("duration_ms", IntegerType),
+        Property("energy", NumberType),
+        Property("id", StringType),
+        Property("instrumentalness", NumberType),
+        Property("key", IntegerType),
+        Property("liveness", NumberType),
+        Property("loudness", NumberType),
+        Property("mode", IntegerType),
+        Property("speechiness", NumberType),
+        Property("tempo", NumberType),
+        Property("time_signature", IntegerType),
+        Property("track_href", StringType),
+        Property("type", StringType),
+        Property("uri", StringType),
+        Property("valence", NumberType),
+    )

tap_spotify/streams.py

@@ -1,11 +1,13 @@
 """Stream type classes for tap-spotify."""
 
 from datetime import datetime
+from typing import Iterable
 
 from singer_sdk.streams.rest import RESTStream
 
 from tap_spotify.client import SpotifyStream
 from tap_spotify.schemas.artist import ArtistObject
+from tap_spotify.schemas.audio_features import AudioFeaturesObject
 from tap_spotify.schemas.track import TrackObject
 from tap_spotify.schemas.utils.rank import Rank
 from tap_spotify.schemas.utils.synced_at import SyncedAt
@@ -36,6 +38,42 @@ def post_process(self, row, context):
         return row
 
 
+class _TracksStream(SpotifyStream):
+    """Define a track stream."""
+
+    def get_records(self, context):
+        # get all track records
+        track_records = list(super().request_records(context))
+
+        # get all audio features records
+        # instantiate audio features stream inline and request records
+        audio_features_stream = _AudioFeaturesStream(self, track_records)
+        audio_features_records = audio_features_stream.request_records(context)
+
+        # merge track and audio features records
+        for track, audio_features in zip(track_records, audio_features_records):
+
+            # account for tracks with `null` audio features
+            row = {**(audio_features or {}), **track}
+            yield self.post_process(row, context)
+
+
+class _AudioFeaturesStream(SpotifyStream):
+    """Define an audio features stream."""
+
+    name = "_audio_features_stream"
+    path = "/audio-features"
+    records_jsonpath = "$.audio_features[*]"
+    schema = AudioFeaturesObject.schema
+
+    def __init__(self, tracks_stream: _TracksStream, track_records: Iterable[dict]):
+        super().__init__(tracks_stream._tap)
+        self._track_records = track_records
+
+    def get_url_params(self, *args, **kwargs):
+        return {"ids": ",".join([track["id"] for track in self._track_records])}
+
+
 class _UserTopItemsStream(_RankStream, _SyncedAtStream, SpotifyStream):
     """Define user top items stream."""
 
@@ -61,11 +99,11 @@ class _UserTopItemsLongTermStream(_UserTopItemsStream):
     time_range = "long_term"
 
 
-class _UserTopTracksStream(_UserTopItemsStream):
+class _UserTopTracksStream(_TracksStream, _UserTopItemsStream):
     """Define user top tracks stream."""
 
     path = "/me/top/tracks"
-    schema = TrackObject.extend_with(Rank, SyncedAt).schema
+    schema = TrackObject.extend_with(Rank, SyncedAt, AudioFeaturesObject).schema
 
 
 class _UserTopArtistsStream(_UserTopItemsStream):
@@ -129,11 +167,11 @@ class UserTopArtistsLongTermStream(
     primary_keys = ["rank", "synced_at"]
 
 
-class _PlaylistTracksStream(_RankStream, _SyncedAtStream, SpotifyStream):
+class _PlaylistTracksStream(_RankStream, _SyncedAtStream, _TracksStream):
     """Define playlist tracks stream."""
 
     records_jsonpath = "$.tracks.items[*].track"
-    schema = TrackObject.extend_with(Rank, SyncedAt).schema
+    schema = TrackObject.extend_with(Rank, SyncedAt, AudioFeaturesObject).schema
     primary_keys = ["rank", "synced_at"]