lockstep-api/api.py

# app.py

# running:
# $ cd /var/sites/api.lockstep.at
# $ source .venv/bin/activate
# $ source .env
# $ flask --app api run --port 8000

# Authomatic 1.3.0 is not compatible with Python 3.12+
# patch: /var/sites/api.lockstep.at/.venv/lib/python3.13/site-packages/authomatic/providers/__init__.py
# def _fetch()
#     HTTPSConnection(
#         # cert_file=cert_file,   # <-- comment this

# sync continuously:
# $ while sleep 1; do diff -q api.py /tmp/api.py; if [ $? -ne 0 ]; then scp api.py lockstep@api.lockstep.at:/var/sites/api.lockstep.at/; cp api.py /tmp/api.py; fi; done

import os
from flask import Flask, request, session, jsonify, make_response
from werkzeug.middleware.proxy_fix import ProxyFix
from authomatic import Authomatic
from authomatic.adapters import WerkzeugAdapter
from authomatic.providers import oauth2
from authomatic.providers import PROVIDER_ID_MAP as AUTHOMATIC_PROVIDER_ID_MAP


class Spotify(oauth2.OAuth2):
    """
    Custom Authomatic provider for Spotify Web API Authorization Code Flow.

    Spotify docs:
      - Authorization endpoint: https://accounts.spotify.com/authorize
      - Token endpoint: https://accounts.spotify.com/api/token
      - Current user profile: https://api.spotify.com/v1/me
    """
    name = "spotify"

    # Provider endpoints
    user_authorization_url = "https://accounts.spotify.com/authorize"
    access_token_url = "https://accounts.spotify.com/api/token"
    user_info_url = "https://api.spotify.com/v1/me"

    # Helpful preset scopes
    spotify_scopes = ["playlist-read-private"]

    def update_user(self):
        """
        Fetch Spotify profile and map it onto Authomatic's generic user object.
        """
        response = self.access(self.user_info_url)

        if response.status != 200:
            return response

        data = response.data or {}

        # Common Authomatic user fields
        self.user.id = data.get("id")
        self.user.name = data.get("display_name") or data.get("id")
        #self.user.email = data.get("email")

        # Spotify has images as a list; use the first one if present.
        images = data.get("images") or []
        if images:
            self.user.picture = images[0].get("url")

        # Optional extra fields if your Authomatic version exposes them
        self.user.username = data.get("id")
        self.user.link = (data.get("external_urls") or {}).get("spotify")

        return response

# "must exist in the same module as Spotify (Authomatic provider class)"
PROVIDER_ID_MAP = list(AUTHOMATIC_PROVIDER_ID_MAP) + [Spotify]


SPOTIFY_CLIENT_ID = os.environ["SPOTIFY_CLIENT_ID"]
SPOTIFY_CLIENT_SECRET = os.environ["SPOTIFY_CLIENT_SECRET"]

# Must exactly match a Redirect URI configured in your Spotify app settings.
REDIRECT_URI = os.environ.get(
    "SPOTIFY_REDIRECT_URI",
    #"https://api.lockstep.at/spotify/callback"
    "https://api.lockstep.at/login/spotify/"
)

app = Flask(__name__)
app.secret_key = os.environ["FLASK_SECRET_KEY"]
app.wsgi_app = ProxyFix(
    app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1
)

CONFIG = {
    "spotify": {
        "class_": Spotify,
        "consumer_key": SPOTIFY_CLIENT_ID,
        "consumer_secret": SPOTIFY_CLIENT_SECRET,

        # Spotify scopes are space-separated conceptually; Authomatic accepts a list.
        "scope": Spotify.spotify_scopes

        # Optional: force Spotify to show consent every time.
        # "user_authorization_params": {"show_dialog": "true"},
    }
}

authomatic = Authomatic(CONFIG, app.secret_key)


@app.route("/")
def index():
    return """
    <h1>Spotify + Authomatic example</h1>
    <p><a href="/login/spotify/">Log in with Spotify</a></p>
    <p>After login, this app fetches your Spotify profile and one extra API call.</p>
    """


@app.route("/login/<provider_name>/", methods=["GET", "POST"])
def login(provider_name):
    response = make_response()

    # Let Authomatic handle the OAuth2 handshake.
    result = authomatic.login(
        WerkzeugAdapter(request, response),
        provider_name,
        session=session,
        session_saver=lambda: session.modified
    )

    # If result is None, Authomatic is still redirecting/processing.
    if result is None:
        return response

    if result.error:
        return jsonify({
            "ok": False,
            "stage": "auth",
            "error": getattr(result.error, "message", str(result.error))
        }), 400

    # Raw parsed JSON returned by Spotify token endpoint
    raw_token_data = None
    raw_token_body = None
    token_http_status = None

    if getattr(result.provider, "access_token_response", None):
        token_http_status = result.provider.access_token_response.status
        raw_token_data = result.provider.access_token_response.data
        raw_token_body = result.provider.access_token_response.content

    if not result.user:
        return jsonify({
            "ok": False,
            "error": "No user returned from Spotify"
        }), 400

    # Fetch Spotify user profile from /v1/me
    result.user.update()

    if result.user.credentials is None:
        return jsonify({
            "ok": False,
            "error": "Login succeeded, but no credentials were returned."
        }), 400

    # Example protected-resource call using the OAuth access token:
    # Spotify current user's top tracks (requires user-top-read scope if enabled)
    # For a scope-free example, use /v1/me only, which we already used in update().
    me_response = result.provider.access("https://api.spotify.com/v1/me")

    payload = {
        "ok": True,
        "provider": provider_name,
        "user": {
            "id": result.user.id,
            "name": result.user.name,
            "email": result.user.email,
            "picture": getattr(result.user, "picture", None),
            "profile_url": getattr(result.user, "link", None),
        },
        "credentials": {
            # Keep this server-side in real apps. Returned here only as a demo.
            "access_token": result.user.credentials.token,
            "refresh_token": getattr(result.user.credentials, "refresh_token", None),
            "expires": str(getattr(result.user.credentials, "expires", None)),
            "token_http_status": token_http_status,
            "raw_token_data": raw_token_data,
        },
        "spotify_me_status": me_response.status,
        "spotify_me_data": me_response.data,
    }

    return jsonify(payload)

"""
{
  "credentials": {
    "access_token": "__TOKEN__",
    "expires": "None",
    "raw_token_data": {
      "access_token": "__TOKEN__",
      "expires_in": 3600,
      "refresh_token": "__TOKEN2__",
      "scope": "playlist-read-private",
      "token_type": "Bearer"
    },
    "refresh_token": "__TOKEN2__",
    "token_http_status": 200
  },
  "ok": true,
  "provider": "spotify",
  "spotify_me_data": {
    "display_name": "cidermole",
    "external_urls": {
      "spotify": "https://open.spotify.com/user/cidermole"
    },
    "followers": {
      "href": null,
      "total": 2
    },
    "href": "https://api.spotify.com/v1/users/cidermole",
    "id": "cidermole",
    "images": [],
    "type": "user",
    "uri": "spotify:user:cidermole"
  },
  "spotify_me_status": 200,
  "user": {
    "email": null,
    "id": "cidermole",
    "name": "cidermole",
    "picture": null,
    "profile_url": "https://open.spotify.com/user/cidermole"
  }
}
"""

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8000, debug=True)