in supporting-blog-content/spotify-to-elasticsearch/python/services.py [0:0]
def get_tracks_metadata(self, track_ids: str) -> Dict:
metadatas = {}
to_fetch = []
for track_id in track_ids:
if track_id["spotify_track_uri"] is not None and track_id[
"spotify_track_uri"
].startswith("spotify:track:"):
track_id = track_id["spotify_track_uri"].replace("spotify:track:", "")
if self.metadata_cache.cache.get(track_id, None) is not None:
metadatas[track_id] = self.metadata_cache.cache[track_id]
else:
to_fetch.append(track_id)
if len(to_fetch) > 0:
spotify_answer = self.client.tracks(to_fetch)
if spotify_answer["tracks"] is not None:
# Spotify can be a bit annoying and send back an Array that has `None` in it.
spotify_answer["tracks"] = [
t for t in spotify_answer["tracks"] if t is not None
]
if len(to_fetch) != len(spotify_answer["tracks"]):
for missing_id in set(to_fetch) - {
t["id"] for t in spotify_answer["tracks"]
}:
console.print(
f"[red] Could not fetch metadata for track id: {missing_id}"
)
if len(spotify_answer["tracks"]) > 0:
for track in spotify_answer["tracks"]:
metadatas[track["id"]] = track
self.metadata_cache.cache[track["id"]] = track
else:
console.print(
f"[red] Could not fetch metadata from Spotify. {to_fetch}"
)
return metadatas