Skip to content

Python

Python Examples

Complete Python examples for integrating with the OpenProspect API using the requests library.

Installation

Install required packages:

Bash
pip install requests python-dotenv

Setup

Environment Variables

Create a .env file:

Bash
# .env
OPENPROSPECT_API_KEY=lnc_live_your_api_key_here
PROSPECT_SEARCH_ID=550e8400-e29b-41d4-a716-446655440000

Load in your code:

Python
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("OPENPROSPECT_API_KEY")
PROSPECT_SEARCH_ID = os.getenv("PROSPECT_SEARCH_ID")
BASE_URL = "https://api.openprospect.io"

API Client Class

Create a reusable client:

Python
import requests
from typing import Dict, Any, Optional, List


class OpenProspectClient:
    """Client for the OpenProspect API."""

    def __init__(self, api_key: str, base_url: str = "https://api.openprospect.io"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Accept": "application/json"
        })

    def _request(
        self,
        method: str,
        endpoint: str,
        params: Optional[Dict[str, Any]] = None,
        json: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """Make HTTP request to API."""
        url = f"{self.base_url}{endpoint}"
        response = self.session.request(
            method=method,
            url=url,
            params=params,
            json=json
        )
        response.raise_for_status()
        return response.json()

    def list_companies(
        self,
        prospect_search_id: str,
        limit: int = 50,
        offset: int = 0,
        min_match_score: Optional[float] = None,
        review_status: Optional[str] = None,
        search: Optional[str] = None
    ) -> Dict[str, Any]:
        """List companies from a prospect search."""
        params = {
            "prospect_search_id": prospect_search_id,
            "limit": limit,
            "offset": offset
        }
        if min_match_score is not None:
            params["min_match_score"] = min_match_score
        if review_status:
            params["review_status"] = review_status
        if search:
            params["search"] = search

        return self._request("GET", "/api/v1/companies/", params=params)

    def get_company(self, company_id: str) -> Dict[str, Any]:
        """Get a single company by ID."""
        return self._request("GET", f"/api/v1/companies/{company_id}")

    def get_company_jobs(
        self,
        company_id: str,
        days_back: int = 30,
        platform: Optional[str] = None,
        limit: int = 50,
        offset: int = 0
    ) -> Dict[str, Any]:
        """Get job listings for a company."""
        params = {
            "days_back": days_back,
            "limit": limit,
            "offset": offset
        }
        if platform:
            params["platform"] = platform

        return self._request("GET", f"/api/v1/companies/{company_id}/jobs", params=params)

    def get_company_jobs_summary(
        self,
        company_id: str,
        days_back: int = 30
    ) -> Dict[str, Any]:
        """Get job activity summary for a company."""
        return self._request(
            "GET",
            f"/api/v1/companies/{company_id}/jobs/summary",
            params={"days_back": days_back}
        )

    def get_company_webtech(
        self,
        company_id: str,
        category: Optional[str] = None,
        min_confidence: int = 0,
        limit: int = 100,
        offset: int = 0
    ) -> Dict[str, Any]:
        """Get technology stack details for a company."""
        params = {
            "min_confidence": min_confidence,
            "limit": limit,
            "offset": offset
        }
        if category:
            params["category"] = category

        return self._request(
            "GET",
            f"/api/v1/companies/{company_id}/webtech-intel",
            params=params
        )

    def get_company_webtech_summary(self, company_id: str) -> Dict[str, Any]:
        """Get technology stack summary for a company."""
        return self._request("GET", f"/api/v1/companies/{company_id}/webtech-intel/summary")

Basic Examples

List Companies

Python
from openprospect_client import OpenProspectClient

client = OpenProspectClient(API_KEY)

# Get first 50 companies
response = client.list_companies(
    prospect_search_id=PROSPECT_SEARCH_ID,
    limit=50
)

print(f"Total companies: {response['total']}")
print(f"Retrieved: {len(response['items'])}")
print(f"Has more: {response['has_more']}")

for company in response['items']:
    print(f"- {company['name']} (Match: {company['match_score']})")

Filter Companies

Python
# Get high-quality accepted companies
response = client.list_companies(
    prospect_search_id=PROSPECT_SEARCH_ID,
    min_match_score=7.0,
    review_status="accepted",
    limit=20
)

for company in response['items']:
    print(f"{company['name']} - {company['city']} - Score: {company['match_score']}")

Text Search

Python
# Search for software companies
response = client.list_companies(
    prospect_search_id=PROSPECT_SEARCH_ID,
    search="software",
    limit=25
)

print(f"Found {response['total']} companies matching 'software'")

Get Single Company

Python
company_id = "7f3e8c90-1234-5678-9abc-def012345678"
company = client.get_company(company_id)

print(f"Company: {company['name']}")
print(f"Website: {company['website']}")
print(f"Match Score: {company['match_score']}")
print(f"Contacts: {len(company['contacts'])}")

for contact in company['contacts']:
    print(f"  - {contact['name']} ({contact['title']}): {contact['email']}")

Pagination

Paginate Through All Results

Python
def get_all_companies(
    client: OpenProspectClient,
    prospect_search_id: str,
    **filters
) -> List[Dict[str, Any]]:
    """Fetch all companies with pagination."""
    all_companies = []
    offset = 0
    limit = 100  # Max per request

    while True:
        response = client.list_companies(
            prospect_search_id=prospect_search_id,
            limit=limit,
            offset=offset,
            **filters
        )

        all_companies.extend(response['items'])
        print(f"Fetched {len(all_companies)}/{response['total']} companies...")

        if not response['has_more']:
            break

        offset += limit

    return all_companies


# Usage
all_companies = get_all_companies(
    client,
    PROSPECT_SEARCH_ID,
    min_match_score=7.0,
    review_status="accepted"
)

print(f"Total companies retrieved: {len(all_companies)}")

Pagination with Progress Bar

Python
from tqdm import tqdm

def get_all_companies_with_progress(
    client: OpenProspectClient,
    prospect_search_id: str,
    **filters
) -> List[Dict[str, Any]]:
    """Fetch all companies with progress bar."""
    # Get total count first
    first_response = client.list_companies(
        prospect_search_id=prospect_search_id,
        limit=1,
        **filters
    )
    total = first_response['total']

    all_companies = []
    offset = 0
    limit = 100

    with tqdm(total=total, desc="Fetching companies") as pbar:
        while offset < total:
            response = client.list_companies(
                prospect_search_id=prospect_search_id,
                limit=limit,
                offset=offset,
                **filters
            )

            all_companies.extend(response['items'])
            pbar.update(len(response['items']))

            offset += limit

    return all_companies

Job Listings

Get Company Jobs

Python
company_id = "7f3e8c90-1234-5678-9abc-def012345678"

# Get all jobs from last 30 days
jobs_response = client.get_company_jobs(
    company_id=company_id,
    days_back=30,
    limit=50
)

print(f"Total jobs: {jobs_response['total']}")
print(f"Activity: {jobs_response['activity_strength']}")
print(f"Velocity: {jobs_response['velocity_per_month']} jobs/month")

for job in jobs_response['jobs']:
    print(f"- {job['title']} on {job['platform']} (posted {job['date_posted']})")

Filter Jobs by Platform

Python
# Get only LinkedIn jobs
linkedin_jobs = client.get_company_jobs(
    company_id=company_id,
    platform="linkedin",
    days_back=60
)

print(f"LinkedIn jobs: {linkedin_jobs['total']}")

Get Job Summary

Python
summary = client.get_company_jobs_summary(company_id=company_id)

print(f"Total jobs: {summary['total']}")
print(f"Activity: {summary['activity_strength']}")
print(f"Velocity: {summary['velocity_per_month']} jobs/month")

if summary['latest_job']:
    job = summary['latest_job']
    print(f"Latest job: {job['title']} on {job['platform']}")

Technology Stack

Get All Technologies

Python
company_id = "7f3e8c90-1234-5678-9abc-def012345678"

webtech = client.get_company_webtech(
    company_id=company_id,
    min_confidence=80
)

print(f"Total technologies: {webtech['total']}")
print(f"Scan successful: {webtech['scan_successful']}")

for tech in webtech['technologies']:
    version = f" ({tech['version']})" if tech.get('version') else ""
    print(f"- {tech['name']}{version} - {tech['category']} ({tech['confidence']}% confidence)")

Filter by Category

Python
# Get only JavaScript frameworks
js_frameworks = client.get_company_webtech(
    company_id=company_id,
    category="javascript_framework",
    min_confidence=90
)

print("JavaScript Frameworks:")
for tech in js_frameworks['technologies']:
    print(f"- {tech['name']} ({tech['confidence']}%)")

Get Technology Summary

Python
summary = client.get_company_webtech_summary(company_id=company_id)

print(f"Total technologies: {summary['total_technologies']}")
print("\nTop technologies:")
for tech in summary['top_technologies']:
    print(f"- {tech['name']} ({tech['category']})")

print("\nCategory breakdown:")
for category, count in summary['category_counts'].items():
    print(f"- {category}: {count}")

Error Handling

Basic Error Handling

Python
import requests

try:
    response = client.list_companies(
        prospect_search_id=PROSPECT_SEARCH_ID,
        limit=50
    )
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 401:
        print("Authentication failed - check your API key")
    elif e.response.status_code == 403:
        print("Insufficient permissions - check your API key scopes")
    elif e.response.status_code == 404:
        print("Prospect search not found")
    elif e.response.status_code == 429:
        print("Rate limit exceeded - wait before retrying")
    else:
        print(f"HTTP error: {e}")
except requests.exceptions.RequestException as e:
    print(f"Request failed: {e}")

Retry Logic with Exponential Backoff

Python
import time
from requests.exceptions import HTTPError


def retry_request(func, max_retries: int = 3):
    """Retry request with exponential backoff."""
    for attempt in range(max_retries):
        try:
            return func()
        except HTTPError as e:
            if e.response.status_code == 429:
                # Rate limit - wait and retry
                wait_time = 2 ** attempt
                print(f"Rate limited. Waiting {wait_time}s before retry...")
                time.sleep(wait_time)
            elif e.response.status_code >= 500:
                # Server error - retry
                wait_time = 2 ** attempt
                print(f"Server error. Retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                # Client error - don't retry
                raise

    raise Exception(f"Failed after {max_retries} retries")


# Usage
response = retry_request(
    lambda: client.list_companies(PROSPECT_SEARCH_ID, limit=50)
)

Data Processing

Export to CSV

Python
import csv

companies = get_all_companies(
    client,
    PROSPECT_SEARCH_ID,
    min_match_score=7.0,
    review_status="accepted"
)

with open('companies.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.DictWriter(f, fieldnames=[
        'name', 'city', 'country', 'website', 'match_score', 'business_type'
    ])
    writer.writeheader()

    for company in companies:
        writer.writerow({
            'name': company['name'],
            'city': company.get('city', ''),
            'country': company.get('country', ''),
            'website': company.get('website', ''),
            'match_score': company['match_score'],
            'business_type': company.get('business_type', '')
        })

print(f"Exported {len(companies)} companies to companies.csv")

Export to JSON

Python
import json

companies = get_all_companies(client, PROSPECT_SEARCH_ID)

with open('companies.json', 'w', encoding='utf-8') as f:
    json.dump(companies, f, indent=2, ensure_ascii=False)

print(f"Exported {len(companies)} companies to companies.json")

Rate Limit Handling

Check Rate Limit Status

Python
response = client.session.get(f"{client.base_url}/health")
print(f"Rate limit: {response.headers.get('X-RateLimit-Limit')}")
print(f"Remaining: {response.headers.get('X-RateLimit-Remaining')}")
print(f"Reset: {response.headers.get('X-RateLimit-Reset')}")

Rate Limit Aware Pagination

Python
import time

def get_all_companies_rate_limited(
    client: OpenProspectClient,
    prospect_search_id: str,
    **filters
) -> List[Dict[str, Any]]:
    """Fetch all companies with rate limit awareness."""
    all_companies = []
    offset = 0
    limit = 100

    while True:
        response = client.list_companies(
            prospect_search_id=prospect_search_id,
            limit=limit,
            offset=offset,
            **filters
        )

        all_companies.extend(response['items'])

        if not response['has_more']:
            break

        offset += limit

        # Rate limiting: 1 request per second
        time.sleep(1)

    return all_companies

Next Steps