Spotify API (top tracks and now playing)

Mix.install([
  {:req, "~> 0.3.10"},
  {:kino, "~> 0.9.4"}
])

Secrets

spotify_client_id = System.fetch_env!("LB_SPOTIFY_CLIENT_ID")
spotify_client_secret = System.fetch_env!("LB_SPOTIFY_CLIENT_SECRET")
spotify_redirect_uri = System.fetch_env!("LB_SPOTIFY_REDIRECT_URI")
:ok

Spotify module

defmodule SpotifyAPI do
  @scopes ["user-top-read", "user-read-currently-playing"]

  def get_authorization_code(client_id, redirect_uri) do
    Req.get!(
      "https://accounts.spotify.com/authorize",
      params: [
        response_type: "code",
        client_id: client_id,
        redirect_uri: redirect_uri,
        scope: URI.encode(Enum.join(@scopes, " "))
      ],
      follow_redirects: false
    )
  end

  def get_access_token(code, client_id, client_secret, redirect_uri) do
    encoded_token = Base.encode64("#{client_id}:#{client_secret}")

    Req.post!(
      "https://accounts.spotify.com/api/token",
      headers: [
        {"Content-Type", "application/x-www-form-urlencoded"},
        {"Authorization", "Basic #{encoded_token}"}
      ],
      body:
        URI.encode_query(
          grant_type: "authorization_code",
          code: code,
          redirect_uri: redirect_uri
        )
    )
  end

  def refresh_access_token(client_id, client_secret, refresh_token) do
    encoded_token = Base.encode64("#{client_id}:#{client_secret}")

    Req.post!("https://accounts.spotify.com/api/token",
      body: URI.encode_query(grant_type: "refresh_token", refresh_token: refresh_token),
      headers: [
        {"Content-Type", "application/x-www-form-urlencoded"},
        {"Authorization", "Basic #{encoded_token}"}
      ]
    )
  end

  def now_playing(access_token) do
    Req.get!(
      "https://api.spotify.com/v1/me/player/currently-playing",
      headers: [{"Authorization", "Bearer #{access_token}"}]
    )
  end

  def top_tracks(access_token) do
    Req.get!(
      "https://api.spotify.com/v1/me/top/tracks",
      params: [limit: 10, offset: 0],
      headers: [{"Authorization", "Bearer #{access_token}"}]
    )
  end
end

defmodule Spotify do
  def process_currently_playing(track) do
    %{
      name: track["name"],
      album_image_url: get_image_url(track),
      album_name: track["album"]["name"],
      artists: get_artists_name(track)
    }
  end

  def process_top_tracks(tracks) do
    Enum.map(tracks, fn track ->
      %{
        name: track["name"],
        album_image_url: get_image_url(track),
        album_name: track["album"]["name"],
        artists: get_artists_name(track)
      }
    end)
  end

  defp get_image_url(track) do
    images = track["album"]["images"]
    first_image = Enum.at(images, 0)
    first_image["url"]
  end

  defp get_artists_name(track) do
    artists = track["artists"]
    Enum.map(artists, fn artist -> artist["name"] end)
  end
end

Getting Authorization Code

Going to use Authorization Flow to get code

response = SpotifyAPI.get_authorization_code(spotify_client_id, spotify_redirect_uri)

if response.status != 303,
  do: raise("Got response status: #{response.status}")

response_headers = response.headers

{"location", redirect_uri} =
  Enum.find(response_headers, fn {key, _value} -> key == "location" end)

click_html = """


  #{redirect_uri}">Go to consent page

"""

Kino.HTML.new(click_html)

Code input

After granting the consent, you will be redirected to redirect_uri with ?code=code. Set the code below.

code_input = Kino.Input.text("Code")

code = Kino.Input.read(code_input)
:ok

Getting auth token and refresh token

response =
  SpotifyAPI.get_access_token(
    code,
    spotify_client_id,
    spotify_client_secret,
    spotify_redirect_uri
  )

if response.status != 200 do
  IO.puts(response.status)
  IO.puts(inspect(response.body))
  raise("Request failed")
end

token_data = response.body

:ok

Now playing

response = SpotifyAPI.now_playing(token_data["access_token"])

now_playing =
  case response.status do
    200 ->
      {:now_playing, response.body}

    204 ->
      {:not_playing, "Nothing playing at the moment"}

    401 ->
      {:access_token_expired}

    _ ->
      IO.puts(response.status)
      {:error, response}
  end

:ok

Processing now playing

case now_playing do
  {:now_playing, now_playing_response} ->
    Spotify.process_currently_playing(now_playing_response["item"])

  {:not_playing, message} ->
    message

  {:access_token_expired} ->
    IO.puts("Access token expired")

  {:error, error_response} ->
    error_response
end

Top songs

response = SpotifyAPI.top_tracks(token_data["access_token"])

top_tracks_response =
  case response.status do
    200 ->
      {:ok, response.body}

    401 ->
      {:access_token_expired}

    _ ->
      IO.puts(response.status)
      {:error, response}
  end

:ok

Processing top songs

case top_tracks_response do
  {:ok, tracks} -> Spotify.process_top_tracks(tracks["items"])
  {:access_token_expired} -> IO.puts("Access token expired")
  {:error, error_response} -> error_response
end

Refresh access token

refresh_token_response =
  SpotifyAPI.refresh_access_token(
    spotify_client_id,
    spotify_client_secret,
    token_data["refresh_token"]
  )