If my REST api has an "api/authenticate" endpoint,...
# singer-tap-development
i
If my REST api has an "api/authenticate" endpoint, which accepts ClientId / username / password parameters to generate a key, how do I declare/ reference these within my custom tap? I see that the sdk APIKeyAuthenticator is imported in my client.py, but how do I use it? I configured the tap to use the "API Key" in the cookiecutter template creation. I've created a python script to generate a token for the API before, can I just plug that logic in somewhere? Or does the authenticator use the name / password in the meltano.yml or something?
p
@Ian OLeary in your client.py that got generated you should see a method called
authenticator
that returns a
APIKeyAuthenticator
(if you selected that in the cookiecutter). It accepts key/value/location parameters see the init method. You just need to pass in your values in:
Copy code
APIKeyAuthenticator.create_for_stream(
            self,
            key=<your key>,
            value=<your value>,
            location=<header or params>,
        )
So if you add your script for generating the token to your stream class you can call it just before the authenticator is initialized here and pass in the api key it returned as the
key
argument. Or you could call it in your init method and store it as an instance variable like
self._auth_value = XYZ
so its not called multiple times when that property is accessed
Does that make sense?
i
class TapNameAuthenticator(OAuthJWTAuthenticator):
"""Authenticator class for JobDiva."""
@classmethod
def create_for_stream(
cls,
stream,  # noqa: ANN001
) -> TapNameAuthenticator:
"""Instantiate an authenticator for a specific Singer stream.
Args:
stream: The Singer stream instance.
Returns:
A new authenticator.
"""
return cls(
stream=stream,
auth_endpoint="<https://api.com/api/authenticate>",
oauth_scopes="TODO: OAuth Scopes",
)
So this is my JWT auth class in my auth.py. would I add my authenticator here or in my client.py auth property which imports and calls this method?
I re-scaffolded a tap since I found out the API uses JWT instead
p
Yep you need to call the create_for_stream method of your custom authenticator class from the stream (client.py)
authenticator
method
i
So once I create the authentication script, If I want to store credentials for the api tap in my .env or just as environment variables when i deploy, how do I manage the testing of the tap with said credentials before I meltano install, and how do I configure that in my meltano.yml
p
Check out https://github.com/meltano/sdk/blob/main/samples/sample_tap_google_analytics/ga_tap_stream.py for an example of a JWT authenticator. You might not need to override the
create_for_stream
method if the SDK has your method covered.
From the example you shared,
Copy code
json={
                "clientid": clientid,
                "username": username,
                "password": password
                }
Can be replace by overrideing this method and returning that dict
oauth_request_payload.
client_id and client_secret are accessible using
self.client_secret
if theyre included as tap configs
Copy code
@property
    def auth_endpoint(self) -> str:
        return "<https://foo.bar.xyz/authenticate>"
@Edgar Ramírez (Arch.dev) you might have more insight into the authenticators honestly. Do you know of other good examples to share?
👀 1
i
@Pat Nadolny (Arch) So I'm still pretty confused and haven't gotten anywhere since yesterday. Is the "create_for_stream" method supposed to return the token for the stream itself? Going off of what Edgar said, I used "self.config" to get the values from the meltano.yml, and declared the properties in the config json in tap.py. My authenticator class now looks like this:
Copy code
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(
                self=self,
                stream=stream,
                auth_token=auth_token
            )
        else:
            raise Exception(f"Failed to authenticate with Tap API: {response.text}")
but I'm still getting the error: TypeError: JobDivaAuthenticator.create_for_stream() missing 1 required positional argument: 'stream'. The good thing, however, is that it does appear to correctly set the clientid, username, and password as the values passed in from the meltano.yml. Am I just doing this completely wrong?
e
Is the "create_for_stream" method supposed to return the token for the stream itself?
No, it should return an instance of
TapAuthenticator
missing 1 required positional argument: 'stream'.
how does the
authenticator
look like in your stream/client class?
i
Copy code
class JobDivaStream(RESTStream):

@cached_property
def authenticator(self) -> _Auth:
        """Return a new authenticator object.

        Returns:
            An authenticator instance.
        """
        return MyTapAuthenticator.create_for_stream(self)
this is what it looks like in my client class