Ian OLeary
01/30/2024, 2:54 PMPat Nadolny (Arch)
01/31/2024, 1:03 PMauthenticator
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:
APIKeyAuthenticator.create_for_stream(
self,
key=<your key>,
value=<your value>,
location=<header or params>,
)
Pat Nadolny (Arch)
01/31/2024, 1:08 PMkey
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 accessedPat Nadolny (Arch)
01/31/2024, 1:08 PMIan OLeary
01/31/2024, 2:12 PMclass 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?Ian OLeary
01/31/2024, 2:12 PMPat Nadolny (Arch)
01/31/2024, 2:31 PMauthenticator
methodIan OLeary
01/31/2024, 4:13 PMPat Nadolny (Arch)
02/07/2024, 6:17 PMcreate_for_stream
method if the SDK has your method covered.Pat Nadolny (Arch)
02/07/2024, 6:20 PMjson={
"clientid": clientid,
"username": username,
"password": password
}
Can be replace by overrideing this method and returning that dict oauth_request_payload.
Pat Nadolny (Arch)
02/07/2024, 6:22 PMself.client_secret
if theyre included as tap configsPat Nadolny (Arch)
02/07/2024, 6:23 PM@property
def auth_endpoint(self) -> str:
return "<https://foo.bar.xyz/authenticate>"
Pat Nadolny (Arch)
02/07/2024, 6:25 PMPat Nadolny (Arch)
02/07/2024, 6:26 PMIan OLeary
02/08/2024, 3:10 PMclass 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?Edgar RamÃrez (Arch.dev)
02/08/2024, 4:20 PMIs 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?Ian OLeary
02/08/2024, 4:30 PMclass 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