lobstrio-sdk 0.7.0

user = client.me()           # User profile
balance = client.balance()   # Account balance (credits, subscription)

crawlers = client.crawlers.list()              # All crawlers
crawler = client.crawlers.get("crawler-id")    # Single crawler
params = client.crawlers.params("crawler-id")  # Parameter schema
attrs = client.crawlers.attributes("crawler-id")  # Result columns

# Does this crawler need an account, and which type?
if crawler.account_type:
    print(f"Needs a {crawler.account_type} account")  # e.g. "sales-nav-sync"

Models: Crawler (includes account: bool and account_type: str | None — two LinkedIn crawlers can need different account types, e.g. linkedin-sync vs sales-nav-sync, so match on account_type, not on the crawler's name), CrawlerAttribute, CrawlerParams

# List & iterate
squids = client.squids.list(limit=50, page=1)
for squid in client.squids.iter():     # auto-paginate all squids
    print(squid.name)

# CRUD
squid = client.squids.create("crawler-id", name="My Project")
squid = client.squids.get("squid-id")
squid = client.squids.update("squid-id", name="Renamed", concurrency=2,
                              params={"language": "English"})
client.squids.empty("squid-id")        # remove all tasks
client.squids.delete("squid-id")

# Link an account-backed crawler's squid to an account
# `accounts` on update() is full-replace on the API side (it detaches
# anything not in the list you send). attach_accounts() reads the squid's
# current accounts first and merges, so it never silently drops one:
squid = client.squids.attach_accounts("squid-id", ["account-hash"])
# Pass replace=True to send exactly the given list (e.g. [] to detach all):
client.squids.attach_accounts("squid-id", [], replace=True)

Model: Squid (id, name, crawler, is_active, concurrency, params, accounts, created_at, ...)

# List & iterate
tasks = client.tasks.list(squid="squid-id")
for task in client.tasks.iter(squid="squid-id"):
    print(task.id)

# Add tasks
result = client.tasks.add(
    squid="squid-id",
    tasks=[
        {"url": "https://maps.google.com/maps?cid=123"},
        {"url": "https://maps.google.com/maps?cid=456"},
    ],
)
print(f"Added {len(result.tasks)}, {result.duplicated_count} duplicates")

# Upload from CSV/TSV
resp = client.tasks.upload(squid="squid-id", file="tasks.csv")
status = client.tasks.upload_status(resp["id"])

# Get & delete
task = client.tasks.get("task-hash")
client.tasks.delete("task-hash")

Models: Task, TaskStatus, AddTasksResult, UploadStatus, UploadMeta

# Start a run
run = client.runs.start(squid="squid-id")

# List runs
runs = client.runs.list(squid="squid-id")
for run in client.runs.iter(squid="squid-id"):
    print(run.id, run.status)

# Monitor
run = client.runs.get("run-id")
stats = client.runs.stats("run-id")
print(f"{stats.percent_done}% done, {stats.total_results} results")

# Wait for completion (blocking, with optional progress callback)
final = client.runs.wait("run-id", poll_interval=5.0,
                          callback=lambda s: print(f"{s.percent_done}%"))

# Start and wait in one call (mirrors start() + wait())
final = client.runs.call(squid="squid-id", poll_interval=5.0,
                         callback=lambda s: print(f"{s.percent_done}%"))

# Bound the wait: RunTimeout (a TimeoutError subclass) if it overruns.
# The run keeps going server-side — re-attach later with wait()/get().
from lobstrio import RunTimeout
try:
    final = client.runs.wait("run-id", timeout=600)   # also on call()
except RunTimeout:
    ...

# Download results
url = client.runs.download_url("run-id")   # signed S3 URL
client.runs.download("run-id", "output.csv")  # download to file

# Abort
client.runs.abort("run-id")

# Tasks within a run
tasks = client.runs.tasks("run-id")

Models: Run, RunStats

results = client.results.list(squid="squid-id", page_size=100)

# Auto-paginate all results
for row in client.results.iter(squid="squid-id"):
    print(row)  # dict

Results are returned as plain dict objects (the schema depends on the crawler).

accounts = client.accounts.list()
account = client.accounts.get("account-id")
types = client.accounts.types()     # available account types

# Sync account with cookies
resp = client.accounts.sync(type="google", cookies={"SID": "...", "HSID": "..."})
status = client.accounts.sync_status(resp["id"])

# Update limits
client.accounts.update("account-id", type="google", params={"daily_limit": 100})

# Delete
client.accounts.delete("account-id")

To link an account to a squid, see client.squids.attach_accounts() above — there is no attach call on AccountsResource itself; the API only accepts accounts on POST /squids/{id}.

Models: Account (id, username, type, status, status_code_info, resets_in, lock_time, squids, ...), AccountType, SyncStatus

# Email
client.delivery.email("squid-id", email="you@example.com")
client.delivery.test_email(email="you@example.com")

# Google Sheets
client.delivery.google_sheet("squid-id", url="https://docs.google.com/spreadsheets/d/...", append=True)
client.delivery.test_google_sheet(url="https://docs.google.com/spreadsheets/d/...")

# Webhook
client.delivery.webhook("squid-id", url="https://your-server.com/hook",
                         on_done=True, on_error=True)
client.delivery.test_webhook(url="https://your-server.com/hook")

# S3
client.delivery.s3("squid-id", bucket="my-bucket", target_path="scrapes/",
                    aws_access_key="...", aws_secret_key="...")
client.delivery.test_s3(bucket="my-bucket")

# SFTP
client.delivery.sftp("squid-id", host="ftp.example.com", username="user",
                      password="pass", directory="/uploads")
client.delivery.test_sftp(host="ftp.example.com", username="user",
                           password="pass", directory="/uploads")

Models: EmailDelivery, GoogleSheetDelivery, S3Delivery, WebhookDelivery, SFTPDelivery

Go to Dashboard → API to find your token. It's always available there, pre-generated.

No. The SDK is standalone. However, if you have the CLI configured (lobstr config set-token), the SDK will automatically pick up the token from ~/.config/lobstr/config.toml — no code changes needed.

Catch RateLimitError and use its retry_after attribute:

from lobstrio import RateLimitError
import time

try:
    results = client.results.list(squid="squid-id")
except RateLimitError as e:
    time.sleep(float(e.retry_after or 5))
    results = client.results.list(squid="squid-id")

Yes. Use AsyncLobstrClient in any async context:

from lobstrio import AsyncLobstrClient

async def scrape_view(request):
    async with AsyncLobstrClient() as client:
        results = await client.results.list(squid="squid-id")
        return results