Save B2B Leads to Postgres with Python and Apify

Bulk lead generation without managing browser clusters

Modern outbound pipeline building requires clean, structured professional data, but building a resilient LinkedIn and search engine extractor from scratch is a massive engineering sink. Scraping public profiles triggers immediate rate limits, requires complex proxy rotation, and demands constant maintenance as DOM selectors shift.

Instead of building and maintaining a custom scraper, we can orchestrate a production-ready solution using the Apify platform. This guide walks through setting up an automated pipeline that invokes the crawlerbros/lead-finder Actor, safely handles API timeouts, pages through the resulting dataset, cleans the records, and upserts them into a PostgreSQL database.

We will build a pipeline designed to run reliably in an asynchronous background worker. By the end of this article, you will have a complete, runnable script that handles network failures, respects API boundaries, and ensures duplicate leads are updated rather than duplicated in your database.

Checked against the Actor's input schema and Apify docs on 2026-09-11.

How to configure the Lead Finder input payload?

You configure the Lead Finder input by constructing a JSON payload with a required jobTitles array and optional arrays for locations, industries, and company names. You must pass these arguments explicitly inside an input JSON payload because the Apify platform prefill values in the Console UI are completely ignored during direct API calls. Only the default schema values apply unless you explicitly define them.

When constructing your payload, the jobTitles field is the only strictly required parameter, and it accepts an array of strings. Each job title in the array triggers a separate search query behind the scenes. If you leave locations, industries, or companyNames empty, the Actor will search globally across all industries and companies. The maxLeads parameter controls the safety limit for the run, capping the number of returned records to prevent unexpected resource utilization.

Here is a complete, production-ready JSON input configuration targeting engineering and product leadership within specific parameters:

{
  "jobTitles": [
    "VP of Engineering",
    "Chief Technology Officer"
  ],
  "locations": [
    "Austin",
    "San Francisco"
  ],
  "industries": [
    "SaaS",
    "Healthcare"
  ],
  "companyNames": [
    "Stripe",
    "Google"
  ],
  "maxLeads": 150
}

What is the output schema of a recovered B2B lead?

Each record returned by the Lead Finder Actor represents a single professional profile enriched with contact details and metadata. The output is structured as a flat JSON object within the dataset, providing clean fields for direct integration.

The output contains key identification fields including the full name, broken down into first and last name components, along with their current job title and employer details. Crucially, the Actor returns the linkedin_url as a stable identifier and a guessed email address constructed from public search patterns. A scraped_at ISO 8601 timestamp is also included, which is vital for handling data freshness and resolving merge conflicts during database upserts.

Here is an example of a single clean record returned by the Actor's dataset endpoint:

{
  "full_name": "Greg Brauner",
  "first_name": "Greg",
  "last_name": "Brauner",
  "title": "VP of Marketing",
  "company_name": "Thinkific",
  "company_domain": "thinkific.com",
  "location": "Austin",
  "linkedin_url": "https://www.linkedin.com/in/gregbrauner",
  "email": "[e-mail]",
  "scraped_at": "2026-02-27T10:18:02.866080+00:00"
}

How to run the Actor asynchronously to avoid timeouts?

You run the Actor asynchronously by making a POST request to its runs endpoint and polling its execution status. Any run that exceeds 300 seconds on a synchronous endpoint is instantly aborted with an HTTP 408 timeout error, making asynchronous polling mandatory for large lead targets. This method returns immediate execution metadata and allows your process to check in at controlled intervals.

To run the Lead Finder Actor reliably, issue an HTTP POST request to the asynchronous run endpoint /v2/acts/crawlerbros~lead-finder/runs. This endpoint immediately returns an execution metadata block containing a unique run ID and status. Your worker process must then poll the execution status endpoint at a reasonable interval until the status field transitions to SUCCEEDED.

Here is a complete Python script using the standard requests library to trigger an asynchronous run, track its progress, and handle errors:

import time
import requests

def run_lead_finder_async(api_token: str, input_data: dict) -> str:
    url = "https://api.apify.com/v2/acts/crawlerbros~lead-finder/runs"
    headers = {"Authorization": f"Bearer {api_token}"}

    # Trigger the asynchronous run
    response = requests.post(url, json=input_data, headers=headers)
    if response.status_code != 201:
        raise RuntimeError(f"Failed to start Actor: {response.text}")

    run_info = response.json()
    run_id = run_info["data"]["id"]
    status_url = f"https://api.apify.com/v2/runs/{run_id}"

    print(f"Started run {run_id}. Polling status...")

    # Poll until execution finishes, fails, or aborts
    while True:
        status_response = requests.get(status_url, headers=headers)
        if status_response.status_code != 200:
            raise RuntimeError(f"Failed to fetch run status: {status_response.text}")

        status_data = status_response.json()["data"]
        status = status_data["status"]

        print(f"Current run status: {status}")

        if status == "SUCCEEDED":
            return status_data["defaultDatasetId"]
        elif status in ["FAILED", "ABORTED", "TIMED-OUT"]:
            raise RuntimeError(f"Actor run ended with terminal status: {status}")

        time.sleep(15)

How to page through large lead datasets without hitting limits?

To retrieve items from large Apify datasets without hitting rate limits, page through the results using the limit and offset query parameters on the items endpoint. The default Apify dataset API has a strict rate limit of 400 requests per second for dataset operations, and loading thousands of large records in a single memory-heavy request can cause network instability.

Using pagination allows you to control memory utilization on your application server and process records in predictable chunks. In Python, you can implement a standard while loop that increments the offset by your page size on each iteration. The loop terminates when the API returns an empty array, signifying that all records have been consumed.

Here is how to implement a paging generator in Python to stream records from the default dataset:

from typing import Generator, Dict, Any
import requests

def page_dataset_items(
    api_token: str, 
    dataset_id: str, 
    batch_size: int = 1000
) -> Generator[Dict[str, Any], None, None]:
    url = f"https://api.apify.com/v2/datasets/{dataset_id}/items"
    headers = {"Authorization": f"Bearer {api_token}"}
    offset = 0

    while True:
        params = {
            "limit": batch_size,
            "offset": offset,
            "clean": "true"
        }

        response = requests.get(url, headers=headers, params=params)
        if response.status_code != 200:
            raise RuntimeError(f"Dataset fetch failed: {response.text}")

        items = response.json()
        if not items or len(items) == 0:
            break

        for item in items:
            yield item

        offset += len(items)

How to save cleaned leads to Postgres with an upsert logic?

To save lead data safely, use an INSERT ON CONFLICT statement to upsert records into your Postgres database using the candidate's LinkedIn URL as the unique constraint. This prevents duplication of contacts when running the Actor multiple times across overlapping industries or titles, ensuring your CRM data remains clean.

First, your database table must have a unique index or primary key constraint on the natural identifier, which is the linkedin_url in the case of lead scrapers. Since emails are guessed and can occasionally be empty, a null-friendly social URL is the most stable key. When a conflict occurs on that LinkedIn URL, the update block refreshes the contact details and updates the tracking timestamps.

Below is the complete database setup script and the Python function using psycopg2 to batch-insert your cleaned leads:

import psycopg2
from psycopg2.extras import execute_values
from typing import List, Dict, Any

# Execute this once to prepare your target database table
CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS b2b_leads (
    linkedin_url VARCHAR(512) PRIMARY KEY,
    full_name VARCHAR(256),
    first_name VARCHAR(128),
    last_name VARCHAR(128),
    title VARCHAR(256),
    company_name VARCHAR(256),
    company_domain VARCHAR(256),
    location VARCHAR(256),
    email VARCHAR(256),
    scraped_at TIMESTAMP WITH TIME ZONE,
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
"""

def upsert_leads_to_postgres(connection_string: str, leads: List[Dict[str, Any]]):
    upsert_query = """
        INSERT INTO b2b_leads (
            linkedin_url, full_name, first_name, last_name, 
            title, company_name, company_domain, location, email, scraped_at
        ) VALUES %s
        ON CONFLICT (linkedin_url) DO UPDATE SET
            full_name = EXCLUDED.full_name,
            first_name = EXCLUDED.first_name,
            last_name = EXCLUDED.last_name,
            title = EXCLUDED.title,
            company_name = EXCLUDED.company_name,
            company_domain = EXCLUDED.company_domain,
            location = EXCLUDED.location,
            email = COALESCE(EXCLUDED.email, b2b_leads.email),
            scraped_at = EXCLUDED.scraped_at,
            updated_at = CURRENT_TIMESTAMP;
    """

    # Map raw JSON fields to database tuple structure, handling missing fields safely
    data_tuples = []
    for lead in leads:
        # Require a valid linkedin_url to proceed
        if not lead.get("linkedin_url"):
            continue

        data_tuples.append((
            lead.get("linkedin_url"),
            lead.get("full_name"),
            lead.get("first_name"),
            lead.get("last_name"),
            lead.get("title"),
            lead.get("company_name"),
            lead.get("company_domain"),
            lead.get("location"),
            lead.get("email"),
            lead.get("scraped_at")
        ))

    if not data_tuples:
        return

    with psycopg2.connect(connection_string) as conn:
        with conn.cursor() as cur:
            # Efficient bulk upserting inside a transaction block
            execute_values(cur, upsert_query, data_tuples)
            conn.commit()

Platform limitations and runtime failure modes

The Lead Finder Actor has specific platform constraints that developers must design around. It relies heavily on public search engine scraping, making it vulnerable to localized rate limiting and search layout changes.

Here are the key limitations and technical failure modes you must account for in production:

• Storage Expiration: On the Apify Free plan, unnamed datasets and run storages are subject to strict deletion rules. Only the 10 most recent runs are retained, and they expire after 4 months. If your backend worker delays downloading the data, the files will be purged. To protect your data, name your runs or extract the datasets immediately upon completion. Named storages are always exempt from deletion.
• Request Queue Processing: A request queue on Apify can only be processed by one Actor or task run at a time. Trying to fan out a single scraper job by starting multiple concurrent runs reading from the same shared request queue will fail or cause unexpected locking. Each concurrent chunk must have its own isolated input and queue.
• Guessed Email Quality: The email output parameter is a guessed pattern based on domain formats found in public search results. It is not verified against SMTP servers. Your pipeline must route these emails through a validation tool to avoid high bounce rates that can damage your cold outreach domain's reputation.
• Proxy Expiration: Residential proxies used to bypass search engine search blocks have a strict session duration limit. On Apify, residential proxy sessions automatically expire after approximately 30 minutes. If your Actor run takes longer due to a large maxLeads limit, session recycling will occur mid-run, which can occasionally cause temporary network retries.

Handling the residential proxy expiration failure mode

Because search engine scraping relies heavily on residential proxies, runs that search for broad terms and have a high maxLeads value can easily cross the 30-minute residential proxy session limit. When a session expires, the underlying connection drops, and the Actor is forced to cycle to a new session. If this occurs mid-request, the Actor might emit network timeout errors or fail to retrieve the next page of search results.

To make your pipeline resilient to this failure mode, you can write Python code that monitors the state of your run. If the run fails due to proxy drops or timeouts, you can leverage Apify's resurrection capabilities. Resurrecting a run restarts the container with the same storage, excludes downtime from duration, and restarts the timeout clock.

Here is a Python function that demonstrates how to catch failures and issue a resurrection request to recover the run:

def handle_run_resurrection(api_token: str, run_id: str) -> str:
    url = f"https://api.apify.com/v2/runs/{run_id}"
    resurrect_url = f"https://api.apify.com/v2/runs/{run_id}/resurrect"
    headers = {"Authorization": f"Bearer {api_token}"}

    status_response = requests.get(url, headers=headers)
    if status_response.status_code != 200:
        raise RuntimeError("Failed to check run status.")

    status_data = status_response.json()["data"]
    status = status_data["status"]

    # If the run aborted or failed, we attempt to resurrect it
    if status in ["FAILED", "ABORTED", "TIMED-OUT"]:
        print(f"Run {run_id} terminated with status {status}. Resurrecting...")
        res_response = requests.post(resurrect_url, headers=headers)

        if res_response.status_code == 200:
            print("Run successfully resurrected. Resume status checking.")
            return "RESURRECTED"
        else:
            raise RuntimeError(f"Resurrection failed: {res_response.text}")

    return status

How the Apify platform billing model scales

The platform costs associated with running this Actor are determined by your selected plan, memory configuration, and proxy bandwidth usage. Understanding how these factors interact is critical to predicting your monthly API bill.

Compute charges and memory selection

Every run on the Apify platform is billed based on Compute Units (CUs). The consumption is calculated using a standard formula:

CU = (memory_mb / 1024) * duration_hours

Running an Actor with 1024MB of memory for exactly one hour consumes 1 CU. Doubling memory is CU-neutral only for autoscaling runs that can scale their resource usage up and down dynamically. However, autoscaling only applies to solutions running multiple tasks or URLs for at least 30 seconds each. Because the Lead Finder run executes sequential search operations, selecting higher memory values directly scales your CU consumption per minute.

The cost per CU depends entirely on your subscription tier. Below are the rates per CU across the four platform tiers:

• Free Plan ($0/mo): CU rate is $0.20
• Starter Plan ($19/mo): CU rate is $0.20
• Scale Plan ($199/mo): CU rate is $0.16
• Business Plan ($999/mo): CU rate is $0.13

Proxy bandwidth consumption

Because search engines aggressively block datacenter IP addresses, this Actor routes its traffic through residential proxies. Residential proxy traffic is billed per gigabyte (GB) of data transferred. The rates per gigabyte decrease as you upgrade to higher plan tiers:

• Free Plan: Residential proxies cost $8 per GB
• Starter Plan: Residential proxies cost $8 per GB
• Scale Plan: Residential proxies cost $7.50 per GB
• Business Plan: Residential proxies cost $7 per GB

Cost management using charge caps

To protect your budget against unexpected recursive loops or massive search tasks, you can limit the financial exposure of any single run using charge caps. When triggering a run via the API, you can pass the maxTotalChargeUsd query parameter. This is automatically used to enforce limits and is exposed inside the running container as the ACTOR_MAX_TOTAL_CHARGE_USD environment variable.

When the Actor's proxy and compute usage approaches this dollar cap, the run is terminated. However, it is important to note that this is not an instant, hard kill. The container keeps consuming resources briefly during its shutdown sequence, meaning the final charge might slightly exceed the exact cap value. Setting this parameter is highly recommended for automated cron jobs and large target pipelines.

The Actor's README is the source of truth for its inputs, outputs and limits. Need a hand wiring this into your stack? Email [e-mail]