Using Constant Contact API

Question

I'm trying to use Python and the Constant Contact API to push new contacts to Constant Contact and add them to a list. So far, I have tried the following:

import requests

url = 'https://api.cc.email/v3/contacts'
url_completed = url + '?api_key=' + api_key

post_request = {
    "email_address": {
        "address": "abc123@gmail.com"
    },
    "first_name": "First",
    "last_name": "Last",
    "create_source": "Account",
    "list_memberships": ["xxxxxxxxx"]
}

headers = {
    'Authorization': 'Bearer xxxxxxx',
    'content_type': 'application/json'
}

resp = requests.post(url_completed, data=post_request, headers=headers)
print(resp.raise_for_status())

But when I run this, I get an error 401 Client Error: Unauthorized for url: https://api.cc.email/v3/contacts?api_key=xxxxxxx

Why am I getting this error? For a little more background just to make sure I did this correctly, I went into the Constant Contact dev portal and created a new application. It gave me an API key, which I inserted into the URL, and a Secret, which I assumed was the bearer token, so substituted it in there. I also tested the payload using the Constant Contact reference documentation UI which allows for test payloads after OAuth authentication, and those succeeded. Specifically, I am using this endpoint.

401: Unauthorized Example error message: The access token used is invalid. Description: The access token or API key that you provided is not valid or has expired. Action: Please provide a valid access token or use the OAuth flow to generate a new token. This is from the website "constantcontact.com". I tried to create an account to see if I can solve the problem but it required to subscribe by money. — , Jul 02 '21 at 17:13
As an aside: `resp.raise_for_status()` raises HttpError on an error and thus there is nothing for your to print. If the contact already exists you will get a 409 status code, which will cause an exception be raised. But om this "error" you probably want to make a new call to update the existing contact instead. So first test `resp.status_code` for a 409 value before you call `raise_for_status`. — Booboo, Mar 07 '23 at 17:58

Booboo · Answer 1 · 2023-03-19T19:09:10.237

As has been mentioned in comments and answers, the 401 code signifies that your token is not valid. This could definitely happen with a once-valid token that has now expired (they have a limited lifetime).

When you received the response for your initial request to get an access token, see if that response contained a refresh token value as key refresh_token. If so, presumably you have saved this away. If not, start the process of obtaining a new access token but this time save the refresh token. When your access token has expired the refresh token can be used to obtain a new access token. So before you call raise_for_status, you should first test for "errors" that are possibly recoverable. Read the comments in the following code:

import requests

url = 'https://api.cc.email/v3/contacts'
# I don't believe you send the api_key here:
#url_completed = url + '?api_key=' + api_key

post_request = {
    "email_address": {
        "address": "abc123@gmail.com"
    },
    "first_name": "First",
    "last_name": "Last",
    "create_source": "Account",
    "list_memberships": ["xxxxxxxxx"]
}

access_token = 'xxxxxxx'

headers = {
    'Authorization': f'Bearer {access_token}',
    # Correct header:
    'Content-Type': 'application/json'
}

# data= changed to json=:
resp = requests.post(url, json=post_request, headers=headers)
if resp.status != 201
    # contact was not successfully added
    if resp.status_code == 401:
        # possible expired token
        access_token = get_new_access_token()
        # Try again with the new access_token
        headers['Authorization'] = f'Bearer {access_token}'
        resp = requests.post(url, json=post_request, headers=headers)

    # contact already exists?
    if resp.status_code != '409':
        # No, raise an exception
        resp.raise_for_status()

    # Here to update existing client.
    # But this requires knowing the contact_id, which is not
    # returned as a documented value. And you can't get it by fetching
    # the contact for which you already need to know the contact_id.
    # Who designed this API? However, the contact_id is
    # buried in an error message (at least today it is -- who knows about
    # tomorrow?).
    import re

    result = resp.json()
    # Look for contact_id
    for item in result:
        m = re.search(r'\b[a-f0-9]+(?:-[a-f0-9]+)+\b', item['error_message'])
        if m:
            break
    if not m:
        raise Exception('Cannot locate contact_id')
     contact_id = m[0]

    # Now attempt to do an update:
    resp = requests.put(f'{url}/{contact_id}', json=post_request, headers=headers)
    resp.raise_for_status()

def get_new_access_token():
    refresh_token = 'yyyyyyy'
    client_id = 'zzzzzzz'
 
    post_data = {'refresh_token': refresh_token, 'grant_type': 'refresh_token', 'client_id': client_id}
    url = 'https://authz.constantcontact.com/oauth2/default/v1/token'
    headers = {
        'Accept': 'application/json',
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    resp = requests.post(url, data=post_data, headers=headers)
    resp.raise_for_status()
    result = resp.json()
    access_token = result['access_token']
    if 'refresh_token' in result:
        refesh_token = result['refresh_token']
    # Save new access_token and refresh_token somewhere:
    ...
    return access_token

score -1 · Answer 2 · answered Jul 07 '21 at 04:22

-1

401 means 'access token is invalid'. Most probably, you must use not the API key provided by the app, but a dedicated access token which is given by OAuth 2.0 flow. Read more about OAuth authorization here.

answered Jul 07 '21 at 04:22

s0mbre

361
2
14

Using Constant Contact API

2 Answers2