LeadIQ Public API Guide

Overview

The LeadIQ API allows you to query professional information about people and companies. You can find a single person, a single company, a list of people or a list of people grouped by company. You can also make utility calls like viewing your API credits or submitting a data correction patch.

---

About GraphQL

The LeadIQ API uses GraphQL, an industry standard approach for APIs which return structured data. Using GraphQL, clients can control exactly which data should be returned by the query, get multiple resources in a single query, and have strong typing of values.

---

API access

To gain access to the LeadIQ API, you'll need an API key. Your API key can be found within your LeadIQ account in Settings > API Keys. For the examples that follow, choose the Secret Base64 API key.

---

Authentication

Authentication to the API is performed via HTTP Basic authentication. Provide your API key as the username value. You do not need to provide a password.

cURL

curl \
  --header "Authorization: Basic $LEADIQ_API_KEY" \
  --header "Content-Type: application/json" \
  --request POST \
  --data '{"query": "query Usage { usage { planUsage { name creditType units cap billingType} subscription {status}}}"}' \
  https://api.leadiq.com/graphql

Python (using requests)

import json
import os
import requests

endpoint = 'https://api.leadiq.com/graphql'
api_key = os.getenv('LEADIQ_API_KEY')
headers = {
    'Authorization': f'Basic {api_key}',
    'Content-Type': 'application/json'
}
body = '{"query": "query Usage { usage { planUsage { name creditType units cap billingType} subscription {status}}}"}'

response = requests.post(endpoint, headers=headers, data=body)
result = response.json()
print(result)

Python (using gql)

from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
import os

api_key = os.getenv('LEADIQ_API_KEY')
transport = AIOHTTPTransport(url="https://api.leadiq.com/graphql", headers={'Authorization': f'Basic {api_key}'})
client = Client(transport=transport, fetch_schema_from_transport=True)

query = gql(
    """
    query Usage {
      usage {
        planUsage {
          name
          creditType
          units
          cap
          billingType
        }
        subscription {
          status
        }
      }
    }
"""
)

result = client.execute(query)
print(result)

---

Credit usage

When calling the LeadIQ API to query professional data, you consume credits. There is a query available to view your API credits. The most common types of credits are:

Credit Type	Used By
Advanced Search (Page)	flatAdvancedSearch or groupedAdvancedSearch
Company (Page)	searchCompany
Contact (Page)	searchPeople
Contact (ExactMatch)	searchPeople if you include personalPhones, workPhones or workEmails in the response data

---

Rate Limits

If the LeadIQ API endpoint receives too many requests associated with the same API key in a short time window, it might respond with a 429 Too Many Requests error.

Note this is not an HTTP status, but rather part of the errors array of the response; see below for guidance on error handling. A best practice when this occurs is to retry your request later with an exponential backoff.

Our default rate limits are 10 requests per minute for free plans and 60 requests per minute for paid plans. If you require a higher rate limit, please contact support at [e-mail].

Plan Level	Allowed Calls Per Minute
Free	10
Paid	60

---

Errors

The LeadIQ API always returns HTTP 200. To detect and handle errors, you therefore cannot rely on HTTP status codes, and you must instead examine the errors array within the JSON response. There are two main types of errors: those affecting the entire request, and those affecting an individual data record.

---

Request errors

Request level errors represent overall problems with the request, and typically result in no data being returned. These can be found in the extensions > response > status field of each error object. Common request level error statuses include:

• 400 - Bad request.
• 401 - API token is incorrect or absent.
• 402 - Insufficient credits.
• 429 - Too many requests.
• 500 - Internal server error.

Sample response

{
  "errors": [
    {
      "message": "401: Unauthorized",
      "extensions": {
        "code": "UNAUTHENTICATED",
        "response": {
          "url": "http://dataiq-web-dataiq-web-chart.dataiq/dataiq/graphql",
          "status": 401,
          "statusText": "Unauthorized",
          "body": "Unauthorized operation Some(Usage__dataiq__0)"
        }
      }
    }
  ],
}

---

Data errors

Data level errors represent problems with individual data records, within a query that potentially returns multiple records. These can be found in the extensions > status field of each error object. Common data level error statuses include:

• 423 - Data obscured for privacy compliance
• No status - Downstream service error

Sample response

{
  "errors": [
    {
      "message": "<Person> information have been obscured for data privacy compliance",
      "path": ['searchPeople'],
      "extensions": {
        "status": 423,
        "code": "FOUND_AND_OBSCURED",
        "trackingRef": 'b4e07e62-fa29-43fc-8de0-26217ed91b6e",
        "serviceName": 'dataiq'
      }
    }
  ]
}

---

Handling errors

A robust error handling approach needs to address three situations: transient errors (e.g. 500) which benefit from immediate retry, throttling errors (e.g. 429) which benefit from retry after a delay, and stable errors (most of the rest) where typically the caller needs to modify the request. See the example below for a full error handling implementation.

---

Example: query person by email

This example uses the LeadIQ API to find a person based on their professional email address. It accepts an input parameter, contains a full error handling loop, and chooses the first position, email and phone number when the API returns a list.

Python (using gql)

from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
from gql.transport.exceptions import TransportQueryError
import json
import os
import time


def findContactByEmail(email_input):
    # Prepare to call the LeadIQ API (with an input parameter)
    # but do not call it yet -- the actual call will be done
    # below in a loop that supports retries in case of error
    api_key = os.getenv('LEADIQ_API_KEY')
    transport = AIOHTTPTransport(url="https://api.leadiq.com/graphql", headers={'Authorization': f'Basic {api_key}'})
    client = Client(transport=transport, fetch_schema_from_transport=True)
    params = {"input": {"email": email_input}}
    query = gql(
        """
        query SearchPeople($input: SearchPeopleInput!) {
          searchPeople(input: $input) {
            totalResults
            hasMore
            results {
              name {
                first
                fullName
                last
              }
              linkedin {
                linkedinUrl
              }
              personalPhones {
                type
                status
                value
              }
              currentPositions {
                title
                seniority
                function
                emails {
                  type
                  status
                  value
                }
                phones {
                  type
                  status
                  value
                }
                companyInfo {
                  name
                }
              }
            }
          }
        }
    """
    )

    # The LeadIQ API returns a full graph, which has the potential to include
    # multiple people, phone numbers and positions. This function returns a
    # single flat record for the person, which means we'll need to choose
    # the first item whenever we are given more than one.
    person = {
        'linkedinUrl': None,
        'first_name': None,
        'last_name': None,
        'direct_mobile': None,
        'title': None,
        'function': None,
        'seniority': None,
        'company': None,
        'work_email': None,
        'work_phone': None
    }

    # Call the LeadIQ API within a loop to handle errors, which includes:
    # 1. Transient errors where we can retry immediately
    # 2. Too many requests errors we we need an exponential backoff
    # 3. Informational errors where we can proceed normally
    sleep_time = 5
    retries = 0
    MAX_RETRIES = 10
    done = False
    while not done:
        if retries >= MAX_RETRIES:
            print(f'Exiting after 10 unsuccessful retries.')
            return
        try:
            start_time = time.perf_counter()
            if retries == 0:
                print(f'Calling LeadIQ API to find [{email_input}]...', end=' ', flush=True)
            json_response = client.execute(query, variable_values=params)
            end_time = time.perf_counter()
            elapsed_time = end_time - start_time
            print(f"{elapsed_time:0.2f} seconds")
            done = True
        except TransportQueryError as e:
            end_time = time.perf_counter()
            elapsed_time = end_time - start_time
            print(f"{elapsed_time:0.2f} seconds")
            error = e.errors[0]
            # Downstream service error -- retry
            if error['extensions']['code'] == 'DOWNSTREAM_SERVICE_ERROR':
                retries += 1
                print(f'Error: Downstream Service Error. Retrying...', end=' ', flush=True)
                continue
            # Response level errors
            elif 'response' in error['extensions']:
                error_status = error['extensions']['response']['status']
                error_text = error['extensions']['response']['statusText']
                # Too many requests -- retry with exponential backoff
                if error_status == 429:
                    retries += 1
                    print(f'Error 429: Too many API requests. Retrying after {sleep_time} seconds...', end=' ', flush=True)
                    time.sleep(sleep_time)
                    sleep_time = sleep_time * 2
                    continue
                # Internal error -- retry
                elif error_status == 500:
                    retries += 1
                    print(f'Error 500: Internal Server Error. Retrying...', end=' ', flush=True)
                    continue
                # Anything else -- exit
                else:
                    print(f'Error {error_status}: {error_text}! Exiting.')
                    return
            # Record level error -- exit
            else:
                error_status = error['extensions']['status']
                error_text = error['message']
                print(f'Error {error_status}: {error_text}. Exiting.')
                return
    # Process (and flatten) the results of the API call
    try:
        data = json_response
        # No person matched
        if data['searchPeople']['totalResults'] == 0:
            print('No person found.')
            return
        # Person found! (Assume we can use the first match.)
        print(f'Person found, returning data.')
        p = data['searchPeople']['results'][0]
        person['first_name'] = p['name']['first']
        person['last_name'] = p['name']['last']
        if 'linkedin' in p:
            person['linkedinUrl'] = p['linkedin']['linkedinUrl']
        if len(p['personalPhones']) > 0:
            person['direct_mobile'] = p['personalPhones'][0]['value'] # return first mobile
        if len(p['currentPositions']) > 0:
            # use the first position with a work email
            position_index = 0
            if len(p['currentPositions']) > 1:
                for index, pos in enumerate(p['currentPositions']):
                    if len(pos['emails']) > 0:
                        position_index = index
                        break
                print(f"Info: more than 1 position found, using position {position_index}")
            position = p['currentPositions'][position_index]
            person['title'] = position['title']
            person['function'] = position['function']
            person['seniority'] = position['seniority']
            person['company'] = position['companyInfo']['name']
            if len(position['emails']) > 0:
                person['work_email'] = position['emails'][0]['value'] # return first work email
            if len(position['phones']) > 0:
                person['work_phone'] = position['phones'][0]['value'] # return first work phone
    except Exception as e:
        print(f"Error when processing [{email_input}] --> {e} ")
        print(json_response)
        return

    return(person)

If you have any questions about API, you can reach out to the LeadIQ Support Team by clicking Submit a Request at the top right of this page.