[Python] How to generate an authorization code from a given scope and use it to generate a...

I'm trying to integrate with the Zoho CRM API using Python, and I need to generate an authorization code from a given scope (ZohoCRM.modules.ALL) and then use that authorization code to generate a refresh token.

Currently I am generating code from Zoho developer Console and replacing it manually everytime. I have to generate a new code every time when i am calling API.

• How do I automate the process of obtaining the grant code?
• Is it possible to fully automate this without any manual browser interaction?

Here is the code Than I am using.

CLIENT_ID: str = "12345"
CLIENT_SECRET = "12345"
REDIRECT_URI = "https://accounts.zoho.in**"
TOKEN_URL = "https://accounts.zoho.in/oauth/v2/token"
CRM_BASE_URL = "https://www.zohoapis.in/crm/v7"
ORG_ID = "00000"


# Global variable to store refresh_token
refresh_token = None

class ZohoOAuth:
def __init__(self):
# Retrieve sensitive data from environment variables
self.CLIENT_ID: str = config("CLIENT_ID", cast=str)
self.CLIENT_SECRET = config("CLIENT_SECRET", cast=str)
self.REDIRECT_URI = config("REDIRECT_URI", cast=str)
self.SCOPE = "ZohoCRM.modules.ALL"
self.token_url = "https://accounts.zoho.in/oauth/v2/token"

async def exchange_code_for_tokens(self, code: str):
# Exchange authorization code for access and refresh tokens
payload = {
'grant_type': 'authorization_code',
'client_id': self.CLIENT_ID,
'client_secret': self.CLIENT_SECRET,
'redirect_uri': self.REDIRECT_URI,
'code': code,
}
print(payload)
try:
response = requests.post(self.token_url, data=payload)
response.raise_for_status() # Will raise an exception for bad status codes
return response.json() # Return the access and refresh tokens
except requests.exceptions.RequestException as e:
raise HTTPException(status_code=400, detail=f"Error obtaining token: {str(e)}")


@router.get("/generate_token")
async def generate_token(request: Request):
global refresh_token
try:
# Capture authorization code from Zoho redirect
code = "1000.b500427f0cafcfccc0d456b87503413c.cd5e8d693a143f0054df5bbdb4a706b9"

if not code:
raise HTTPException(status_code=400, detail="Authorization code missing")

# Ensure we await the async method
tokens = await ZohoOAuth().exchange_code_for_tokens(code)

if "error" in tokens:
raise HTTPException(status_code=400, detail=f"Zoho API Error: {tokens['error']}")

# Store the refresh_token globally after successful exchange
refresh_token = tokens.get("refresh_token")

return {
"access_token": tokens.get("access_token"),
"refresh_token": refresh_token,
"scope": tokens.get("scope"),
"api_domain": tokens.get("api_domain"),
"token_type": tokens.get("token_type"),
"expires_in": tokens.get("expires_in")
}

except requests.exceptions.RequestException as e:
print(f"Request error: {str(e)}") # Log error
raise HTTPException(status_code=500, detail="Error communicating with Zoho API")

except Exception as e:
print(f"Unexpected error: {str(e)}") # Log error
raise HTTPException(status_code=500, detail="An unexpected error occurred")


# ✅ Function to get a fresh access token using the global refresh token
def get_access_token():
global refresh_token

if not refresh_token:
raise Exception("Refresh token is missing. Please ensure that you have successfully authenticated.")

payload = {
"grant_type": "refresh_token",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"refresh_token": refresh_token,
}
response = requests.post(TOKEN_URL, data=payload)
data = response.json()

if "access_token" in data:
return data["access_token"]
else:
raise Exception(f"Error fetching access token: {data}")

# Create Zoho Module Endpoint
@router.post("/create_account_module")
async def create_account_modules():
"""Create a new custom module in Zoho CRM."""
try:
result = create_account_module()
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

# ✅ Function to fetch the first available profile ID (mandatory for module creation)
def get_profile_id():
from src.api.routes.zohocrm import get_access_token
access_token = get_access_token()
headers = {"Authorization": f"Zoho-oauthtoken {access_token}", "X-CRM-ORG": ORG_ID}

response = requests.get(f"{CRM_BASE_URL}/settings/profiles", headers=headers)
if response.status_code == 200:
profiles = response.json().get("profiles", [])
if profiles:
return profiles[0]["id"] # Use the first profile ID
else:
raise Exception("No profiles found.")
else:
raise Exception(f"Error fetching profiles: {response.json()}")



# ✅ Function to create a custom module "DigitalYou"
def create_account_module():
from src.api.routes.zohocrm import get_access_token
access_token = get_access_token() # Ensure this is synchronously called
profile_id = get_profile_id() # Ensure this is synchronously called
print(access_token)
print(profile_id)

headers = {
"Authorization": f"Zoho-oauthtoken {access_token}",
"X-CRM-ORG": ORG_ID,
"Content-Type": "application/json"
}

module_data = {}

response = requests.post(f"{CRM_BASE_URL}/settings/modules", json=module_data, headers=headers)

if response.status_code == 201:
return response.json() # ✅ Module created successfully
else:
return f"Error: {response.json()}"



After the code used once I am getting error like

Traceback (most recent call last):
File "/home/roshni/Roshni/AI Avatar/B2B/B2B-DY-Backend/src/api/routes/zohocrm.py", line 62, in generate_token
raise HTTPException(status_code=400, detail=f"Zoho API Error: {tokens['error']}")
fastapi.exceptions.HTTPException: 400: Zoho API Error: invalid_code

TypeError: 'coroutine' object is not callable

Continue reading...