are there any good examples of using an authenticator outside of a Stream? as far as I can tell, it requires a stream to create one, but I need to use it as part of discovery. I'm not sure why each stream needs its own instance of an authenticator in the first place. parallel connections maybe?

Not a great working example (tested modifying auth configuration of

tap-spotify

), but you can create a stub stream from the tap yourself for the time being:

auth.py

"""Spotify Authentication."""

from singer_sdk.authenticators import OAuthAuthenticator, SingletonMeta
from singer_sdk.plugin_base import PluginBase as TapBaseClass


class SpotifyAuthenticator(OAuthAuthenticator, metaclass=SingletonMeta):
    """Authenticator class for Spotify."""

    @property
    def oauth_request_body(self):
        return {
            "grant_type": "refresh_token",
            "refresh_token": self.config["refresh_token"],
            "client_id": self.config["client_id"],
            "client_secret": self.config["client_secret"],
        }

    @property
    def auth_endpoint(self) -> str:
        return "<https://accounts.spotify.com/api/token>"

    @classmethod
    def create_for_tap(cls, tap: TapBaseClass):
        class StreamStub:
            tap_name = tap.name
            config = tap.config
            logger = tap.logger

        return cls(StreamStub())

tap.py

"""Spotify tap class."""

from memoization import cached
from singer_sdk import Tap
from singer_sdk import typing as th

from tap_spotify import streams
from tap_spotify.auth import SpotifyAuthenticator

STREAM_TYPES = [
    streams.UserTopTracksShortTermStream,
    streams.UserTopTracksMediumTermStream,
    streams.UserTopTracksLongTermStream,
    streams.UserTopArtistsShortTermStream,
    streams.UserTopArtistsMediumTermStream,
    streams.UserTopArtistsLongTermStream,
    streams.GlobalTopTracksDailyStream,
    streams.GlobalTopTracksWeeklyStream,
    streams.GlobalViralTracksDailyStream,
]


class TapSpotify(Tap):
    """Spotify tap class."""

    name = "tap-spotify"

    config_jsonschema = th.PropertiesList(
        th.Property(
            "client_id",
            th.StringType,
            required=True,
            description="App client ID",
        ),
        th.Property(
            "client_secret",
            th.StringType,
            required=True,
            description="App client secret",
        ),
        th.Property(
            "refresh_token",
            th.StringType,
            required=True,
            description="Refresh token",
        ),
    ).to_dict()

    @property
    @cached
    def authenticator(self) -> SpotifyAuthenticator:
        """Return a new authenticator object."""
        return SpotifyAuthenticator.create_for_tap(self)

    def discover_streams(self):
        return [stream_class(tap=self) for stream_class in STREAM_TYPES]

client.py

"""REST client handling, including SpotifyStream base class."""

from typing import Optional
from urllib.parse import ParseResult, parse_qsl

from memoization import cached
from singer_sdk.streams import RESTStream

from tap_spotify.pagination import BodyLinkPaginator


class SpotifyStream(RESTStream):
    """Spotify stream class."""

    url_base = "<https://api.spotify.com/v1>"
    records_jsonpath = "$.items[*]"

    @property
    @cached
    def authenticator(self):
        return self._tap.authenticator

    def get_new_paginator(self):
        return BodyLinkPaginator()

    def get_url_params(self, context, next_page_token: Optional[ParseResult]):
        params = super().get_url_params(context, next_page_token)
        return dict(parse_qsl(next_page_token.query)) if next_page_token else params

Tap still works with this change:

$ meltano config tap-spotify test
2023-09-06T12:00:39.798276Z [info     ] The default environment 'test' will be ignored for `meltano config`. To configure a specific environment, please use the option `--environment=<environment name>`.
Plugin configuration is valid