does anyone have a working example of their "auth.py" that uses jwt authentication? I'm struggling with getting one working

Hi @Ian OLeary! I don't think I actually have an example. What part are you struggling with?

So, I believe I got the "auth.py" returning a token, but still I'm getting an error "| WARNING | tap-jobdiva | Config validation failed: 'auth_token' is a required property; 'project_ids' is a required property", so do I need to declare somewhere in my meltano.yml that the 'auth_token' should be the cached authenticator property? Or am I misunderstanding. I manually set the "project_ids" to the same project id as "tap-myproject" in my meltano.yml. I'm running this by navigating to

taps\tap-mytap

, running

meltano install

, and then running

meltano invoke tap-mytap

What does your

tap.py

look like? Seems like you have required settings

auth_token

and

project_ids

that are missing?

class TapMyTap(Tap):
    """MyTap tap class."""

    name = "tap-mytap"

    # TODO: Update this section with the actual config values you expect:
    config_jsonschema = th.PropertiesList(
        th.Property(
            "auth_token",
            th.StringType,
            required=True,
            secret=True,  # Flag config as protected.
            description="The token to authenticate against the API service",
        ),
        th.Property(
            "project_ids",
            th.ArrayType(th.StringType),
            required=True,
            description="Project IDs to replicate",
        ),
        th.Property(
            "start_date",
            th.DateTimeType,
            description="The earliest record date to sync",
        ),
        th.Property(
            "api_url",
            th.StringType,
            default="<https://api.mytap.com/api>",
            description="The url for the API service",
        ),
    ).to_dict()

    def discover_streams(self) -> list[streams.MyTapStream]:
        """Return a list of discovered streams.

        Returns:
            A list of discovered streams.
        """
        return [
            streams.NewUpdatedCandidateRecordsStream(self),
        ]

I haven't altered them in the tap.py yet. I've defined one stream in my streams.py (for now - for simplicity) which is shown here

seems like you want client_id, username and password as settings and pass those to your authenticator instance

so define them in the auth_token property itself?

so define them in the auth_token property itself?

what would that look like?

rather as their own properties:

config_jsonschema = th.PropertiesList(
        th.Property(
            "client_id",
            required=True,
             ...
        ),
        th.Property(
            "username",
            required=True,
             ...
        ),
        th.Property(
            "password",
            required=True,
             ...
        ),

and then in your auth class:

clientid = self.config["client_id"]
username = self.config["username"]
password = self.config["password"]

Oh so I really don't need an auth_token property, do I?

No, not really

class TapAuthenticator(OAuthJWTAuthenticator):
    """Authenticator class for Tap."""

    @classmethod
    def create_for_stream(
        self,
        cls,
        stream,  # noqa: ANN001
    ) -> TapAuthenticator:

        clientid = self.config["client_id"]
        username = self.config["username"]
        password = self.config["password"]

        api_url = f"<https://api.tap.com/api/authenticate?clientid={clientid}&username={username}&password={password}>"

        response = requests.get(api_url)

        if response.status_code ==200:
            auth_token = response.text
            return cls(
                stream=stream,
                auth_token=auth_token
            )
        else:
            raise Exception(f"Failed to authenticate with Tap API: {response.text}")

So here's what I've got now, Should I also be returning self in return cls()?