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) and its async export
page = client.jobs.expired("2026-08-15", page=1, limit=100)
task = client.jobs.export_expired("2026-08-15", limit=5000)   # free

# Vector search: by free text, or "more like this" from a job id
similar = client.jobs.vsearch(query, search_type="summary", job_id="6958cfd211e2763c3491ef8b", 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"}],
}

# 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)

# 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)

result = client.jobs.search(query, limit=50)
usage = client.last_usage
print(usage.meter, usage.total_used, "/", usage.included_limit, "remaining:", usage.included_remaining)

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)