Hello!
I’m developing a tap and for the API that I...
# singer-tap-developmentj
jean_sahlberg
02/02/2023, 12:19 PMHello!
I’m developing a tap and for the API that I’m using I need to first POST a request, wait for the report to be generated and then GET the data from the API
I already implemented the authenticator method and it is working fine, I can get the data from existing reports, but now I’m implementing the report generation, for that, I want to reuse the authorization token used by the stream, I’m wondering how can I use the APIKeyAuthenticator to run the POST method. Can someone share an example of a tap that does it?
jean_sahlberg
02/02/2023, 12:25 PMthis is part of my
client.py Copy code
@property
def authenticator(self) -> APIKeyAuthenticator:
"""Return a new authenticator object."""
talkdesk_auth = TalkdeskAuth(self.config)
token_request = talkdesk_auth.request_bearer_token()
return APIKeyAuthenticator.create_for_stream(
self,
key="Authorization",
value=f"Bearer {token_request['access_token']}",
location="header"
)
def generate_report(self):
start_time = datetime.utcnow() - timedelta(days=3)
end_time = datetime.utcnow() - timedelta(hours=2)
data = json.dumps({
"format": "json",
"timespan": {
"from": start_time.strftime("%Y-%m-%dT%H:%M:%SZ"),
"to": end_time.strftime("%Y-%m-%dT%H:%M:%SZ")
}
})
headers={'Authorization':'Bearer token',
'Content-Type':'application/json'}
resp = <http://requests.post|requests.post>("<https://api.talkdeskapp.com/data/reports/calls/jobs>", headers=headers, data=data)
datar = resp.json()
return datar['job']['id']generate_reporte
edgar_ramirez_mondragon
02/02/2023, 5:49 PMHi @jean_sahlberg! (Is the report persistent? If so, it might not be efficient to generate on every run of the tap)
This PR could enable such a use case by simply doing
Copy code
self.requests_session.auth = self.authenticator
<http://self.requests_session.post|self.requests_session.post>(...)j
jean_sahlberg
02/02/2023, 6:58 PMhi Edgar, thank you for your reply, I will take a look!