pip install hirebase

pip install "hirebase[streaming]"   # ijson, for streaming JSON-array exports
pip install "hirebase[cli]"         # the bundled `hirebase` command-line tool

import hirebase

client = hirebase.Client(api_key="hb_...")

export HIREBASE_API_KEY="hb_..."
export HIREBASE_BASE_URL="https://api.hirebase.org"   # optional, this is the default

import hirebase

client = hirebase.Client(api_key="hb_...")

# Search jobs — results are typed and iterable
result = client.jobs.search({
    "job_titles": ["Software Engineer", "Product Engineer"],
    "locations": [{"city": "San Francisco", "region": "California",
                   "country": "United States"}],
    "limit": 20,
})

print(result.total_count, "matches")
for job in result:
    print(job.job_title, "@", job.company_name, "—", job.salary_range)

import asyncio
import hirebase

async def main():
    async with hirebase.AsyncClient(api_key="hb_...") as client:
        result = await client.jobs.search({"job_titles": ["Engineer"]})
        for job in result:
            print(job.job_title)

asyncio.run(main())

# Search
result = client.jobs.search(query, page=1, limit=20)

# Fetch one job by id
job = client.jobs.get("6958cfd211e2763c3491ef8b")

# Market insights for a cohort (same filter shape as search)
insights = client.jobs.insights({"job_titles": ["Data Scientist"]})
print(insights.headline.median_salary, insights.salary.p90)

# How many jobs a query would bill (free; same rule exports use)
cost = client.jobs.estimate(query, limit=100)        # -> int

# Expired-jobs feed (identifiers only, costs 0, since within the last 14 days) and its async export
from datetime import date, timedelta
yesterday = (date.today() - timedelta(days=1)).isoformat()
page = client.jobs.expired(yesterday, page=1, limit=100)
task = client.jobs.export_expired(yesterday, limit=5000)   # free

# "More like this" from a job: neural search with the job's slug (see below)
similar = client.jobs.neural_search(company_slug="acme", job_slug="senior-backend-engineer", limit=10)

from hirebase import JobQuery, SalaryRange

query = JobQuery(
    job_titles=["Backend Engineer"],
    salary=SalaryRange(min=150_000, currency="USD"),
    location_types=["Remote"],
    visa=True,
)
result = client.jobs.search(query)

from hirebase import NeuralSearchQuery, NeuralVectorQuery, JobQuery

result = client.jobs.neural_search(
    vector=NeuralVectorQuery(
        query="Senior Python developer with machine learning and cloud infrastructure experience"
    ),
    lexical=JobQuery(
        location_types=["Remote"],
        experience=["Senior"],
        industry=["Tech, Software & IT Services"],
        days_ago=30,
        limit=10,
    ),
)

for job in result.jobs:
    print(job.job_title, "@", job.company_name)

embed = client.resumes.embed("/path/to/resume.pdf")
embedding = embed.result.embedding  # or embed["result"]["embedding"] for dict

matches = client.jobs.neural_search(
    vector=NeuralVectorQuery(vectors=[embedding]),
    lexical=JobQuery(location_types=["Remote"], limit=25),
)

for job in matches.jobs:
    print(job.job_title, "@", job.company_name)

query = {
    "job_titles": ["Software Engineer", "Product Engineer", "Fullstack Engineer"],
    "locations": [{"city": "San Francisco", "region": "California",
                   "country": "United States"}],
    "limit": 500,   # defaults to 10 if omitted; exports bill per job
}

# 1. Start the export -> returns a Task
task = client.jobs.export(query, format="json")   # or format="csv"

# 2. Poll until it finishes -> (success, result)
success, result = client.tasks.poll(task)
if not success:
    raise RuntimeError(f"Export failed: {result.error}")

# 3. Download the file (streamed to disk)
client.stream_file(result["download_url"], file_path="./jobs.json")

# 4. Stream jobs from the file (typed by default; constant memory)
for job in client.jobs.stream_file("./jobs.json"):
    print(job.job_title)

# ...or get raw dicts
for row in client.jobs.stream_file("./jobs.json", return_type=dict):
    ...

for job in client.jobs.stream_url(result["download_url"]):
    print(job.job_title)

filters = {
    "job_titles": ["sustainability", "ESG", "climate"],
    "geo_locations": [{"country": "United States"}],
    "date_posted_from": "2023-01-01",
}

# 1. Discover (free) -> dict with total_count, by_year_posted, top_companies, sample_titles, export_quote
res = client.jobs.historical_search(filters, sample_size=25)
print(res["total_count"], "matches,", res["export_quote"]["price_usd"], "USD to export all")

# 2. Export (paid per record) -> checkout_url + a pre-assigned task_id
order = client.jobs.historical_export(filters, format="csv", limit=5000)
print("Pay here:", order["checkout_url"])

# 3. After payment the task runs; poll it like any export
success, result = client.tasks.poll(order["task_id"], timeout=1800)
client.stream_file(result["download_url"], file_path="./historical.csv")

# Research one posting (LinkedIn profiles only)
task = client.jobs.contacts("6ab32db8f1c329e432f72994")
success, result = client.tasks.poll(task, timeout=600)
for c in result["contacts"]:
    print(c["name"], c["role"], c["profile_url"])

# Reveal one contact you picked (work email by default, phone on request)
reveal = client.jobs.reveal_contact(c["profile_url"], name=c["name"], reveal_phone=True)
success, found = client.tasks.poll(reveal, timeout=300)
print(found["email"], found["phone_number"])

# Or reveal every contact of a posting up front
task = client.jobs.contacts("6ab32db8f1c329e432f72994", reveal_email=True, reveal_phone=True)

# Search
companies = client.companies.search({"company_name": "Stripe"})
for company in companies:
    print(company.company_name, company.company_slug)

# Get a company by slug — optionally with its jobs and live insights
company = client.companies.get("stripe", return_jobs=True, return_insights=True)
print(company.description_summary)
print(company.insights_data.headline.total_count)

# Bound helpers (the object remembers its client)
insights = company.insights()
jobs = company.get_jobs(limit=10)

# Company-scoped insights directly
insights = client.companies.insights("stripe", query={"days_ago": 30})

# Paginated jobs for a company (bills the jobs meter per job returned)
jobs = client.companies.jobs("stripe", page=1, limit=20, sort_by="date_posted", sort_order="desc")

from hirebase import SalaryBenchmarkRequest

task = client.jobs.salary_benchmark(SalaryBenchmarkRequest(
    job_title="Senior Software Engineer",
    yoe_range={"min": 5, "max": 10},
    geo_locations=[{"country": "United States"}],
    days_ago=90,
))
success, report = client.tasks.poll(task, interval=5, timeout=900)
market = report["market"]["salary"]
print(market["count"], "postings, median", market["p50"], "confidence", report["confidence"]["grade"])

record = client.resumes.upload("./resume.pdf")            # stored on Hirebase
record = client.resumes.get(record.id)
parsed = client.resumes.parse(record.id)
record = client.resumes.upload_and_parse("./resume.pdf")  # both in one call

task = client.tasks.get(task_id)                # current state / progress / result
success, result = client.tasks.poll(task)       # block until finished (or timeout)

summary = client.usage.get()                     # -> UsageSummary
jobs = summary.meter("m_jobs_api_calls")         # or client.usage.get(meter=Meters.JOBS_API)
print(jobs.used, "/", jobs.included, "remaining", jobs.remaining, "mode", jobs.overage_mode)

jobs, meta = client.jobs.search(query, limit=50, return_meta=True)
usage = meta.usage                       # UsageSnapshot, or None on un-metered endpoints
print(usage.meter, usage.total_used, "/", usage.included_limit, "remaining:", usage.included_remaining)
if usage.is_meter_mode and usage.overage_used:
    print("billing overage units:", usage.overage_used)
print(meta.status_code, meta.request_id)  # transport details for the same call

import time, hirebase

try:
    client.jobs.search(query, limit=100)
except hirebase.QuotaExceededError as err:
    remaining = err.usage.included_remaining   # e.g. 2 -> retry with limit=2
except hirebase.RateLimitError as err:
    time.sleep(err.retry_after or 5)           # the 100 requests / 60 s limiter

data = client.jobs.search(query, return_type=dict)   # -> dict
job  = client.jobs.get(job_id, return_type=dict)      # -> dict

import hirebase

client = hirebase.Client(api_key="hb_...")

try:
    client.jobs.search(query)
except hirebase.RateLimitError:
    ...
except hirebase.APIError as e:
    print(e.status_code, e.message)