Hello, I am developing a Python interaction with the Hubspot and I saw that I need the OAuth2 access token, for example, the client code and the client secret. Where can I find them?
Hello @JPereiradaSil, I believe this OAuth Quickstart Guide by Hubspot will help you solve all your questions: Working with OAuth | OAuth Quickstart Guide - HubSpot docs
I has the issue this is the code
import jsonimport secretsimport osfrom fastapi import FastAPI,Request, HTTPExceptionfrom fastapi.responses import HTMLResponseimport httpximport asyncioimport base64from dotenv import load_dotenvfrom integrations.integration_item import IntegrationItemfrom redis_client import add_key_value_redis, get_value_redis, delete_key_redis# Load environment variables from .env fileload_dotenv()CLIENT_ID = os.getenv('HUBSPOT_CLIENT_ID')CLIENT_SECRET = os.getenv('HUBSPOT_CLIENT_SECRET')REDIRECT_URI = os.getenv('HUBSPOT_REDIRECT_URI')encoded_client_id_secret = base64.b64encode(f'{CLIENT_ID}:{CLIENT_SECRET}'.encode()).decode()# Define the necessary scopes for HubSpotscope = 'oauth crm.objects.contacts.read'authorization_url = ( f'https://app.hubspot.com/oauth/authorize' f'?client_id={CLIENT_ID}' f'&redirect_uri={REDIRECT_URI}' f'&scope={scope}')async def authorize_hubspot(user_id, org_id): state_data = { 'state': secrets.token_urlsafe(32), 'user_id': user_id, 'org_id': org_id } encoded_state = base64.urlsafe_b64encode(json.dumps(state_data).encode('utf-8')).decode('utf-8') auth_url = f'{authorization_url}&state={encoded_state}' await add_key_value_redis(f'hubspot_state:{org_id}:{user_id}', json.dumps(state_data), expire=600) return auth_urlasync def oauth2callback_hubspot(request: Request): if request.query_params.get('error'😞 raise HTTPException(status_code=400, detail=request.query_params.get('error')) code = request.query_params.get('code') encoded_state = request.query_params.get('state') state_data = json.loads(base64.urlsafe_b64decode(encoded_state).decode('utf-8')) original_state = state_data.get('state') user_id = state_data.get('user_id') org_id = state_data.get('org_id') saved_state = await get_value_redis(f'hubspot_state:{org_id}:{user_id}') if not saved_state or original_state != json.loads(saved_state).get('state'😞 raise HTTPException(status_code=400, detail='State does not match.') token_url = 'https://api.hubapi.com/oauth/v1/token' async with httpx.AsyncClient() as client: response = await client.post( token_url, data={ 'grant_type': 'authorization_code', 'client_id': CLIENT_ID, 'client_secret': CLIENT_SECRET, 'redirect_uri': REDIRECT_URI, 'code': code, }, headers={ 'Content-Type': 'application/x-www-form-urlencoded', } ) if response.status_code != 200: raise HTTPException(status_code=response.status_code, detail='Failed to fetch access token.') token_data = response.json() await add_key_value_redis(f'hubspot_credentials:{org_id}:{user_id}', json.dumps(token_data), expire=600) close_window_script = """ <html> <script> window.close(); </script> </html> """ return HTMLResponse(content=close_window_script)async def get_hubspot_credentials(user_id, org_id): credentials = await get_value_redis(f'hubspot_credentials:{org_id}:{user_id}') if not credentials: raise HTTPException(status_code=400, detail='No credentials found.') credentials = json.loads(credentials) await delete_key_redis(f'hubspot_credentials:{org_id}:{user_id}') return credentialsasync def create_integration_item_metadata_object(response_json): name = response_json.get('name', 'Unknown Name') integration_item_metadata = IntegrationItem( id=response_json.get('id'), type='hubspot_contact', name=name, creation_time=response_json.get('createdAt'), last_modified_time=response_json.get('updatedAt'), parent_id=None, # HubSpot contacts don't have a parent ID in this context ) return integration_item_metadataasync def get_items_hubspot(credentials): credentials = json.loads(credentials) access_token = credentials.get("access_token") async with httpx.AsyncClient() as client: response = await client.get( 'https://api.hubapi.com/crm/v3/objects/contacts', headers={ 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json', } ) if response.status_code == 200: contacts = response.json().get('results', []) list_of_integration_item_metadata = [ await create_integration_item_metadata_object(contact) for contact in contacts ] return list_of_integration_item_metadata else: raise HTTPException(status_code=response.status_code, detail='Error fetching HubSpot contacts')"