Summary
How to integrate Intuit OAuth 2.0 credentials into a custom Python destination check function using requests and OAuth2Session for authentication.
Question
<@U04SE163WC8> How do I integrate the following code into my check function for my custom python destination?
import requests
from requests.auth import HTTPBasicAuth
from requests_oauthlib import OAuth2Session
# Replace these with your actual Intuit OAuth 2.0 credentials
client_id = 'your_client_id'
client_secret = 'your_client_secret'
redirect_uri = 'your_redirect_uri'
authorization_base_url = '<https://appcenter.intuit.com/connect/oauth2>'
token_url = '<https://oauth.platform.intuit.com/oauth2/v1/tokens>'
# Define the OAuth 2.0 session
intuit = OAuth2Session(client_id, redirect_uri=redirect_uri)
# Get the authorization URL
authorization_url, state = intuit.authorization_url(authorization_base_url)
print('Please go to', authorization_url)
authorization_response = input('Paste the full redirect URL here: ')
# Fetch the access token
intuit.fetch_token(token_url, authorization_response=authorization_response, auth=HTTPBasicAuth(client_id, client_secret))
# Now you can make requests using the intuit session
response = intuit.get('<https://sandbox-quickbooks.api.intuit.com/v3/company/><realm_id>/query?query=SELECT * FROM Customer')
print(response.json())
This topic has been created from a Slack thread to give it more visibility.
It will be on Read-Only mode here. Click here if you want to access the original thread.
Join the conversation on Slack
["intuit-oauth-2.0", "python-destination", "requests", "oauth2session"]
Hmm, I don’t know enough to give you a confident answer yet. The extracts provided do not contain specific information on how to integrate the provided code into a check function for a custom Python destination in Airbyte. However, I can guide you on how to create a custom OAuth for your connector based on the information from the forum post. Here’s an example: python from requests.auth import AuthBase from requests_oauthlib import OAuth1 class MyAuth(AuthBase): def __init__(self, oauth): self.oauth = oauth def __call__(self, r): r.headers["Authorization"] = self.oauth return r oauth = OAuth1( client_key=config["consumer_key"], client_secret=config["consumer_secret"], resource_owner_key=config["access_token"], resource_owner_secret=config["token_secret"], ) return [MyStream(authenticator=MyAuth(oauth))] For more detailed guidance, you may want to refer to the Airbyte documentation or ask your question on the #help-connector-development channel on Slack.
<@U04SE163WC8> I want to use Oauth2 session instead
Based on the extracts provided, you can use the OAuth2Authenticator class provided by Airbyte’s CDK for OAuth2.0 authentication. This class places the API token in the Authorization header and has mechanisms to refresh the current access token given a refresh token. Here’s an example of how you might use it: python from airbyte_cdk.sources.streams.http.auth import OAuth2Authenticator # Replace these with your actual OAuth 2.0 credentials client_id = 'your_client_id' client_secret = 'your_client_secret' refresh_token = 'your_refresh_token' token_refresh_endpoint = 'your_token_refresh_endpoint' authenticator = OAuth2Authenticator( token_refresh_endpoint=token_refresh_endpoint, client_id=client_id, client_secret=client_secret, refresh_token=refresh_token, ) # Pass the created authenticator into the relevant HTTPStream constructor stream = MyStream(authenticator=authenticator) Please note that the OAuth2Authenticator currently only supports refresh tokens and not the full OAuth2.0 loop. For more details, you can refer to the Airbyte documentation on HTTP-API-based Connectors.
<@U04SE163WC8> For my this python fucntion
"""
Tests if the input configuration can be used to successfully connect to the destination with the needed permissions
e.g: if a provided API token or password can be used to connect and write to the destination.
:param logger: Logging object to display debug/info/error to the logs
(logs will not be accessible via airbyte UI if they are not passed to this logger)
:param config: Json object containing the configuration of this destination, content of this json is as specified in
the properties of the spec.json file
:return: AirbyteConnectionStatus indicating a Success or Failure
"""
try:
client_id = 'your_client_id'
client_secret = 'your_client_secret'
redirect_uri = 'your_redirect_uri'
authorization_base_url = '<https://appcenter.intuit.com/connect/oauth2>'
token_url = '<https://oauth.platform.intuit.com/oauth2/v1/tokens>'
# Define the OAuth 2.0 session
intuit = OAuth2Session(client_id, redirect_uri=redirect_uri)
# Get the authorization URL
authorization_url, state = intuit.authorization_url(authorization_base_url)
print('Please go to', authorization_url)
authorization_response = input('Paste the full redirect URL here: ')
# Fetch the access token
intuit.fetch_token(token_url, authorization_response=authorization_response, auth=HTTPBasicAuth(client_id, client_secret))
# Now you can make requests using the `intuit` session
response = intuit.get('<https://sandbox-quickbooks.api.intuit.com/v3/company/><realm_id>/query?query=SELECT * FROM Customer')
print(response.json())
return AirbyteConnectionStatus(status=Status.SUCCEEDED)
except Exception as e:
return AirbyteConnectionStatus(status=Status.FAILED, message=f"An exception occurred: {repr(e)}")
{"type": "TRACE", "trace": {"type": "ERROR", "emitted_at": 1704979150790.75, "error": {"message": "Something went wrong in the connector. See the logs for more details.", "internal_message": "'str' object has no attribute 'json'", "stack_trace": "Traceback (most recent call last):\n File \"/workspaces/airbyte/airbyte-integrations/connectors/destination-quickbooks/main.py\", line 11, in <module>\n DestinationQuickbooks().run(sys.argv[1:])\n File \"/workspaces/airbyte/airbyte-integrations/connectors/destination-quickbooks/.venv/lib/python3.10/site-packages/airbyte_cdk/destinations/destination.py\", line 120, in run\n print(message.json(exclude_unset=True))\nAttributeError: 'str' object has no attribute 'json'\n", "failure_type": "system_error"}}}```
The error message you’re seeing is AttributeError: 'str' object has no attribute 'json'. This error is occurring because you’re trying to call the .json() method on a string object, which is not valid because the .json() method is for response objects, not strings. In your code, you’re trying to print the JSON response of an HTTP request with print(response.json()). However, the error suggests that response is a string, not a response object. Here’s the problematic line in your code: python response = intuit.get('<https://sandbox-quickbooks.api.intuit.com/v3/company/><realm_id>/query?query=SELECT * FROM Customer') print(response.json()) You should ensure that the intuit.get() method is returning a response object. If intuit.get() is not returning a response object as expected, you should debug this method to understand why. If intuit.get() is indeed returning a response object, then the issue might be with the URL you’re passing to it. Make sure that the URL is correctly formatted and does not contain any placeholders that haven’t been replaced with actual values (like <realm_id>). If you’re still having trouble, you might want to ask for help on the #help-connector-development channel on Slack.
<@U04SE163WC8> how do I read the credientials in my secrets/config.json file so that I can access them in my check function in my destinations.py? My current secrets.config is something like this
"client_id": "ABR7ZnMNKrZ1rmwIkkD1f5sFy8LErmA4eH2rbru9O9daKhNr3x",
"client_secret": "VhmFgseGtvwtGXXbFRuldVXdb4ZfS89ZoJDpWX2N",
"refresh_token": "AB11713626778Ug8H1pDCubASFLW1U6tzFbTRhgSBiCCVjaBkd",
"access_token": "EBAgdpr7E3QEhEtY3Og6by7q1BpL4EO-_8zaj_ID-.YQSweK6QWN87Lv463y9hLQ",
"realm_id": "4620816365381465210",
"redirect_uri": "<https://developer.intuit.com/v2/OAuth2Playground/RedirectUrl>"
}```
I want to be able to access them in my check function, currently I have used this
```def check(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> AirbyteConnectionStatus:
try:
client_id = 'your_client_id'
client_secret = 'your_client_secret'
redirect_uri = 'your_redirect_uri'
authorization_base_url = '<https://appcenter.intuit.com/connect/oauth2>'
token_url = '<https://oauth.platform.intuit.com/oauth2/v1/tokens>'
# Define the OAuth 2.0 session
intuit = OAuth2Session(client_id, redirect_uri=redirect_uri)
```
Sorry <@U047WETGHFW>, I seem to be having a bit of trouble. Don’t worry though, I’ve pinged my support folks and they will try to fix me up! 
Perhaps try again in 5mins?