Testing contact availability latency via the Cordial API

api-contact-latency-test.py

import json
from urllib.parse import quote
import urllib3
import requests

DEFAULT_API_VERSION = 'v2'
DEFAULT_API_DOMAIN = 'api.cordial.io'
CORDIAL_APIKEY = ''
urllib3.disable_warnings()

def runtime(sec):
    hours = int(sec/3600)
    minutes = int((sec % 3600) / 60)
    seconds = int(sec % 60)
    return "%02dh:%02dm:%02d.%03ds" % (hours, minutes, seconds, (sec - int(sec)))

def add_contact(**kwargs):
    if not kwargs['Email']:
        raise Exception("Email address is required")

    data = {
        'channels': {
            'email': {
                'address': kwargs['Email'],
                'subscribeStatus': 'subscribed'
            }
        }
    }
    if 'Data' in kwargs and kwargs['Data']:
        data = {**data, **kwargs['Data']}

    url = kwargs['ApiUrl'] + 'contacts'
    headers = {'content-type': 'application/json'}
    r = requests.post(url, auth=(CORDIAL_APIKEY, ''), headers=headers, data=json.dumps(data), verify=False)

    return r.json()

def fetch_contact(**kwargs):
    if not kwargs['Email']:
        raise Exception("Email address is required")

    url = kwargs['ApiUrl'] + 'contacts' + '/'
    if args.cidpk:
        url = url + 'email:'
    url = url + quote(kwargs['Email'])
    headers = {'content-type': 'application/json'}
    r = requests.get(url, auth=(CORDIAL_APIKEY, ''), headers=headers, verify=False)

    return r.json()

if __name__ == "__main__":
    import time
    import argparse
    import os

    parser = argparse.ArgumentParser()
    parser.add_argument('email', type=str, help='email address to add/manage')
    parser.add_argument('--api-key', help="api-key for the Cordial API")
    parser.add_argument('--profile', help="profile to use from credentials file")
    parser.add_argument('--data', default="{}", help="JSON object representing additional contact data")
    parser.add_argument('--version', default=DEFAULT_API_VERSION, help="API version setting")
    parser.add_argument('--domain', default=DEFAULT_API_DOMAIN, help="API domain")
    parser.add_argument('--cidpk', default=False, action="store_true", help="Flag for whether or not account uses cID as PK")

    args = parser.parse_args()
    args.url = "https://%s/%s/" % (args.domain, args.version)

    # Determine profile
    profile = args.profile if args.profile \
        else os.environ['CORDIAL_PROFILE'] if 'CORDIAL_PROFILE' in os.environ \
        else 'default'

    # Process credentials
    if not args.api_key:
        if 'CORDIAL_APIKEY' in os.environ:
            args.api_key = os.environ['CORDIAL_APIKEY']
        else:
            # check for ~/.cordial/credentials
            if os.path.isdir(os.path.expanduser('~/.cordial')) and os.path.isfile(os.path.expanduser('~/.cordial/credentials')):
                import configparser
                creds = configparser.ConfigParser()
                creds.read_file(open(os.path.expanduser('~/.cordial/credentials')))
                if creds.has_option(profile, 'apikey'):
                    args.api_key = creds.get(profile, 'apikey')

    if not args.api_key:
        raise Exception("Unable to locate API key")
    CORDIAL_APIKEY = args.api_key
    args.data = json.loads(args.data)

    # Add the contact
    print ("[%s] ADD %s" % (runtime(0), add_contact(Email=args.email, Data=args.data, ApiUrl=args.url)))

    # Fetch the contact
    start = time.time()
    response = fetch_contact(Email=args.email, ApiUrl=args.url)
    while '_id' not in response:
        print("\r[%s] FETCH %s" % (runtime(time.time()-start), (response['_id'] if '_id' in response else json.dumps(response))), end="")
        time.sleep(1)
        response = fetch_contact(Email=args.email, ApiUrl=args.url)
    print("\r[%s] FETCH %s" % (runtime(time.time()-start), (response['_id'] if '_id' in response else json.dumps(response))))

Requires:

• python3.6+
• urllib3 and requests packages (install via pip)

Test by running:

python -u api-contact-latency.py --cidpk --api-key $apiKey $emailAddress

Example output:

[00h:00m:00.000s] ADD {'success': True, 'message': 'contact created'}
[00h:00m:00.000s] FETCH 5eb3051c57xxxxxxxx0f8e7f