Python Examples

These examples use httpx and the public external API.

Setup

export OPENPROSPECT_API_KEY="lnc_live_your_api_key_here"
python -m pip install httpx

Shared Client

import os
from typing import Any

import httpx


class OpenProspectClient:
    def __init__(self, api_key: str) -> None:
        self._client = httpx.Client(
            base_url="https://api.openprospect.io/api/v1",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30.0,
        )

    def validate(self) -> dict[str, Any]:
        response = self._client.get("/auth/validate")
        response.raise_for_status()
        return response.json()

    def create_enrichment_order(self) -> dict[str, Any]:
        response = self._client.post(
            "/orders",
            json={
                "order_type": "ENRICHMENT",
                "title": "Python example enrichment",
                "features": ["COMPANY_DATA", "CONTACTS"],
                "companies": [
                    {
                        "company_name": "Stripe",
                        "external_id": "crm-1001",
                        "website_url": "https://stripe.com",
                    }
                ],
            },
        )
        response.raise_for_status()
        return response.json()

    def get_order_status(self, order_id: str) -> dict[str, Any]:
        response = self._client.get(f"/orders/{order_id}")
        response.raise_for_status()
        return response.json()

    def get_order_results(self, order_id: str) -> dict[str, Any]:
        response = self._client.get(f"/orders/{order_id}/results", params={"limit": 50})
        response.raise_for_status()
        return response.json()

    def list_prospect_searches(self) -> dict[str, Any]:
        response = self._client.get("/prospect-searches", params={"limit": 50})
        response.raise_for_status()
        return response.json()

    def sync_delivered_companies(self, prospect_search_id: str) -> dict[str, Any]:
        response = self._client.get(
            f"/deliveries/{prospect_search_id}/companies",
            params={"limit": 100},
        )
        response.raise_for_status()
        return response.json()

    def close(self) -> None:
        self._client.close()


client = OpenProspectClient(os.environ["OPENPROSPECT_API_KEY"])
try:
    print(client.validate())
    order = client.create_enrichment_order()
    print(order["order_id"])
finally:
    client.close()

Async Client

import os
from typing import Any

import httpx


async def list_orders() -> dict[str, Any]:
    async with httpx.AsyncClient(
        base_url="https://api.openprospect.io/api/v1",
        headers={"Authorization": f"Bearer {os.environ['OPENPROSPECT_API_KEY']}"},
        timeout=30.0,
    ) as client:
        response = await client.get("/orders", params={"limit": 50})
        response.raise_for_status()
        return response.json()