JobsPipe Python SDK

The official Python library for the JobsPipe API.

Every job posting, one API. Search millions of live jobs from 30+ sources (LinkedIn, Y Combinator, company career pages, and more), normalized into one clean schema — plus website tech-stack detection.

• Synchronous and asynchronous clients
• Fully typed responses (pydantic models, ships py.typed)
• Automatic retries with Retry-After support
• Transparent offset pagination (client.jobs.iter(...))
• A typed exception hierarchy you can branch on

Installation

pip install jobspipe

Requires Python 3.8+.

Authentication

Get an API key from your JobsPipe dashboard (it starts with jp_live_). Pass it directly or set the JOBSPIPE_API_KEY environment variable — the client reads it automatically.

export JOBSPIPE_API_KEY="jp_live_..."

Quickstart

from jobspipe import Jobspipe

client = Jobspipe()  # reads JOBSPIPE_API_KEY, or Jobspipe(api_key="jp_live_...")

results = client.jobs.search(
    description_or=["python", "django"],
    job_country_code_or=["US"],
    remote=True,
    posted_at_max_age_days=7,
    limit=25,
    include_total_results=True,
)

print(f"{results.metadata.total_results} matches")
for job in results.data:
    print(f"{job.job_title} @ {job.company} ({job.country_code}) — {job.url}")

Detect a website's tech stack

scan = client.stack.scan("stripe.com")
print(f"{scan.domain} scanned at {scan.scanned_at} (HTTP {scan.http_status})")
for tech in scan.detected:
    print(f"  {tech.name:20} {tech.confidence:>3.0f}%  {', '.join(tech.categories)}")

Pagination

client.jobs.iter(...) walks every page for you and yields individual jobs. It stops automatically when there are no more results, or when max_results is reached.

for job in client.jobs.iter(
    description_or=["rust"],
    remote=True,
    page_size=100,
    max_results=500,
):
    print(job.job_title, "-", job.company)

Prefer manual control? search() accepts offset/page and the response exposes metadata.next_cursor (which is None on the last page).

Async

Every method has an async counterpart on AsyncJobspipe:

import asyncio
from jobspipe import AsyncJobspipe

async def main():
    async with AsyncJobspipe() as client:
        results = await client.jobs.search(remote=True, limit=10)
        async for job in client.jobs.iter(description_or=["go"], max_results=50):
            print(job.job_title)

asyncio.run(main())

Error handling

All errors derive from jobspipe.JobspipeError. HTTP errors are APIStatusError subclasses carrying the status code, request id, and parsed body.

from jobspipe import (
    Jobspipe,
    AuthenticationError,
    PaymentRequiredError,
    RateLimitError,
    APIStatusError,
    APIConnectionError,
)

client = Jobspipe()
try:
    client.jobs.search(limit=10)
except AuthenticationError:
    print("Bad or missing API key")
except PaymentRequiredError:
    print("Monthly quota exceeded — upgrade your plan")
except RateLimitError as e:
    print(f"Rate limited; retry after {e.retry_after}s")
except APIStatusError as e:
    print(f"API error {e.status_code}: {e.message} (request {e.request_id})")
except APIConnectionError:
    print("Could not reach the API")

Retries & timeouts

Transient failures (429, 408, 409, 5xx, and network errors) are retried automatically with exponential backoff, honoring the server's Retry-After header. Configure per-client:

import httpx
from jobspipe import Jobspipe

client = Jobspipe(
    max_retries=4,                              # default 2
    timeout=httpx.Timeout(30.0, connect=5.0),   # default 60s
)

Configuration

The response models allow extra fields, so new attributes added by the API are preserved (accessible via attribute or model_extra) without an SDK upgrade. Likewise, search() forwards any unknown keyword argument as a filter.

Development

pip install -e ".[dev]"
ruff check .
mypy src
pytest

The test suite mocks HTTP with respx — no network or API key required. Set JOBSPIPE_API_KEY to additionally run the live smoke test.

License

MIT — see LICENSE.