Python Examples

These examples use httpx and the public external API.

Setup

export OPENPROSPECT_API_KEY="lnc_live_your_api_key_here"
export OPENPROSPECT_PROFILE_ID="42665897-b189-4e0f-8841-499351a30aff"
export OPENPROSPECT_IDEMPOTENCY_KEY="replace-with-one-persisted-uuid"
python -m pip install httpx

Generate and persist one idempotency key before the first submission. Re-export that exact value when retrying the same request after a timeout or lost response. Generate a new key only for a new logical order.

Shared Client

import os
from collections.abc import Iterator
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,
        profile_id: str,
        idempotency_key: str,
    ) -> dict[str, Any]:
        response = self._client.post(
            "/orders",
            headers={"Idempotency-Key": idempotency_key},
            json={
                "order_type": "ENRICHMENT",
                "title": "Python example enrichment",
                "profile_id": profile_id,
                "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 iter_order_results(self, order_id: str) -> Iterator[dict[str, Any]]:
        offset = 0
        while True:
            response = self._client.get(
                f"/orders/{order_id}/results",
                params={"limit": 100, "offset": offset},
            )
            response.raise_for_status()
            page = response.json()
            items = page["items"]
            yield from items
            if not page["has_more"]:
                break
            offset += len(items)

    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"])
profile_id = os.environ["OPENPROSPECT_PROFILE_ID"]
idempotency_key = os.environ["OPENPROSPECT_IDEMPOTENCY_KEY"]
try:
    print(client.validate())
    order = client.create_enrichment_order(
        profile_id=profile_id,
        idempotency_key=idempotency_key,
    )
    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()