Python SDK

The lantfly package does two jobs:

Actor: the runtime for actors written in Python. It has the same environment, storages, OUTPUT record, error rows, proxy URLs, pay-per-event charges and graceful stop as the Node.js SDK, plus a request queue.
LantflyClient (and LantflyClientAsync): a client for the Lantfly API (/api/v3) that you can use from any Python program.

Python 3.10 or newer. Standard library only: the package has no dependencies. The import name is lantfly.

Install

Coming soon to PyPI

lantfly (version 0.2.1) is not published to PyPI yet. Once it is, install it with the commands below.

pip
requirements.txt

Terminal
pip install lantfly

requirements.txt
lantfly>=0.2,<1

Write an actor

main.py
import asyncio
from lantfly import Actor


async def main() -> None:
    async with Actor:  # init; on exit: flush, OUTPUT, status message, exit 0
        actor_input = await Actor.get_input()  # the INPUT record, {} when there is none
        proxy = Actor.create_proxy_configuration(
            actor_proxy_input=actor_input.get("proxyConfiguration")
        )
        for tag in actor_input.get("hashtags", []):
            target = Actor.target(f"#{tag}")  # one input = one target in OUTPUT
            posts = await crawl(tag, proxy.new_url_sync(tag) if proxy else None)
            await Actor.push_data(posts)
            await target.ok(len(posts))  # or empty() / blocked(code, msg, url) / failed(...)
            await Actor.charge("page_opened")  # pay-per-event (does nothing locally)


asyncio.run(main())

An uncaught exception inside async with Actor: writes OUTPUT with status: "failed" and exits with code 1. await Actor.fail("message") does the same on purpose. await Actor.main(fn) works like Actor.main in the Node.js SDK.

Try it locally

Without LANTFLY_IS_AT_HOME, everything goes to ./storage (or LANTFLY_LOCAL_STORAGE_DIR):

./storage
storage/
  datasets/default/000000001.json
  key_value_stores/default/INPUT.json
  request_queues/default/<id>.json

Put your input in ./storage/key_value_stores/default/INPUT.json and run python main.py. Set LANTFLY_LOG_LEVEL=DEBUG to print debug lines.

Actor API

Area	API
Lifecycle	`async with Actor:` · `await Actor.init()` · `await Actor.exit(message=None, exit_code=0)` · `await Actor.fail(message)` · `await Actor.main(fn, exit=True)`
Input and values	`await Actor.get_input()` · `await Actor.set_value(key, value, content_type=None)` · `await Actor.get_value(key, default_value=None)`
Dataset	`await Actor.push_data(dict or list)` (buffered, sent in batches of 50, flushed on exit) · `await Actor.open_dataset(name=None, id=None)`
Key-value store	`await Actor.open_key_value_store(name=None, id=None)` → `get_value` / `set_value`
Request queue	`await Actor.open_request_queue(name=None, id=None)` → see below
Run outcome	`Actor.target(label)` → `ok(n)`, `empty()`, `blocked(code, msg, url)`, `failed(code, msg, url)`, `not_found()`, `private()`, `restricted()` · `await Actor.push_error(code, message, input=, url=)` · `classify_block(status=, url=, body=, headers=, platform=, expect_json=)`
Proxy	`Actor.create_proxy_configuration(groups=, country=, use_lantfly_proxy=, actor_proxy_input=)` → `new_url_sync(session_id)` / `await new_url(session_id)`; `None` when the run has no proxy
Platform	`await Actor.charge(event_name, count=1)` · `await Actor.set_status_message(msg, is_terminal=False)` · `Actor.is_at_home()` · `Actor.new_client()`
Stop and timeout	`Actor.time_left_secs()` · `Actor.is_aborting()` · `Actor.on("aborting", fn)` / `Actor.off(...)` · `Actor.stop_on_abort(crawler)` · `await Actor.abort(reason)`
Logging	`Actor.log.debug / info / warning / error / exception(message, data=None)`

Values

set_value stores a dict or list as JSON, bytes as is, and a str as text when you give a text/* content type. None deletes the record. get_value returns parsed JSON, text or bytes, or default_value when the record is missing.

Request queue

main.py
queue = await Actor.open_request_queue()
await queue.add_request("https://example.com")
await queue.add_requests([{"url": "https://example.com/a", "userData": {"label": "DETAIL"}}])

while (request := await queue.fetch_next_request()) is not None:
    try:
        ...  # fetch request["url"]
        await queue.mark_request_handled(request)
    except Exception:
        await queue.reclaim_request(request)

Method	Meaning
`add_request(url or dict, forefront=False)` / `add_requests(list, forefront=False)`	Adds requests, de-duplicated by `uniqueKey` (the URL when you give none).
`fetch_next_request()`	The next request, locked for 180 seconds for this client, or `None`.
`mark_request_handled(request)`	Marks it done.
`reclaim_request(request, forefront=False)`	Puts it back for a retry.
`get_request(request_id)`	One request.
`is_empty()` / `is_finished()`	Nothing pending / nothing pending or in progress.
`get_info()`	Counts of the queue.

note

The request queue of a Python actor lives in the run's container (a fresh folder under the temp directory), also on the platform. It does not survive a container restart.

Proxy

create_proxy_configuration(actor_proxy_input=...) takes the proxyConfiguration object of your input (useLantflyProxy, groups, country); explicit arguments win. use_lantfly_proxy=False returns None. new_url_sync("session-1") returns a URL with a sticky session: the same session id keeps the same exit IP.

Run outcome (OUTPUT)

When the actor reported targets or errors, the SDK writes OUTPUT itself: status (ok, partial, empty, blocked, failed), items, errors (code → count), targets (input, status, items, error, message), proxy and finishedAt. It also sets the terminal status message, for example partial: 12 items from 3 targets (BLOCKED×1).

Error rows in the default dataset are {error, errorMessage, input, url} with the codes BLOCKED, LOGIN_REQUIRED, RATE_LIMITED, NOT_FOUND, PRIVATE, AGE_RESTRICTED, EMPTY, SCHEMA_CHANGED, FAILED. An actor that only writes its own OUTPUT and never uses targets keeps it. The rules are the same as in the Node.js SDK.

Graceful stop

On SIGTERM or SIGINT (an abort, or the run's timeout) the SDK emits aborting, flushes the datasets, writes OUTPUT (unfinished targets become failed, "Stopped before finishing") and exits with code 0 within 6 seconds. A second signal exits at once.

stop_before_timeout_secs (30 seconds, at most a quarter of the run) before LANTFLY_TIMEOUT_AT, the SDK emits aborting and flushes, so stop taking work when Actor.is_aborting() is true.

tip

Run blocking work with asyncio.to_thread. The stop handler ends the process even while a worker thread still waits on the network.

Review items

review_item(id=..., rating=..., text=..., author={...}, target={"type": "product", ...}, sellerReply={"text": ..., "at": ...}, publishedAt=..., likes=..., media=[...]) builds a review row in the shared review shape: it sets kind, rounds rating to a whole 1–5 (else None), keeps web media URLs once and keeps a reply only with text. review_target_stats(...) cleans the platform's numbers about the reviewed target.

Environment

The SDK reads the LANTFLY_* variables the platform sets: LANTFLY_IS_AT_HOME, LANTFLY_API_BASE_URL, LANTFLY_TOKEN, LANTFLY_USER_ID, LANTFLY_ACTOR_ID, LANTFLY_ACTOR_RUN_ID, LANTFLY_DEFAULT_DATASET_ID, LANTFLY_DEFAULT_KEY_VALUE_STORE_ID, LANTFLY_DEFAULT_REQUEST_QUEUE_ID, LANTFLY_INPUT_KEY, LANTFLY_MEMORY_MBYTES, LANTFLY_TIMEOUT_AT, LANTFLY_PROXY_HOSTNAME, LANTFLY_PROXY_PORT, LANTFLY_PROXY_PASSWORD, LANTFLY_MAX_PAID_DATASET_ITEMS, LANTFLY_MAX_TOTAL_CHARGE_USD. read_env() returns them as an ActorEnv. Locally you can also set LANTFLY_LOCAL_STORAGE_DIR, LANTFLY_PROXY_URL (a plain upstream proxy; {session} in it becomes the session id) and LANTFLY_LOG_LEVEL. See Actor environment.

Call the API

client.py
import os
from lantfly import LantflyClient

client = LantflyClient(os.environ["LANTFLY_TOKEN"], base_url="https://lantfly.com")  # /api/v3 is added

run = client.actor("lantfly~web-scraper").call(
    run_input={"startUrls": [{"url": "https://example.com"}], "maxPagesPerCrawl": 5}
)
for item in client.dataset(run["default_dataset_id"]).iterate_items():
    print(item)

output = client.key_value_store(run["default_key_value_store_id"]).get_record("OUTPUT")["value"]

Constructor

Argument	Default	Meaning
`token`	`LANTFLY_TOKEN`	API token (Bearer). Inside a run: the run token.
`base_url`	`LANTFLY_API_BASE_URL`	Server URL, with or without `/api/v3`. Pass `https://lantfly.com`.
`account`	none	Organization id: acts for that organization (`X-Lantfly-Account` header).
`max_retries`	`8`	Retries of network errors, 429 and 5xx.
`min_delay_between_retries_millis`	`500`	First backoff delay.
`timeout_secs`	`360`	Timeout of one HTTP call.

Resources

Client	Methods
`actors()` / `actor(id)`	`list(my=, search=, page=, per_page=)` · `get` · `update` · `delete` · `start(run_input=, build=, memory_mb=, timeout_secs=, max_items=, max_total_charge_usd=, wait_for_finish=)` · `call(..., wait_secs=)` · `build(version=, git_url=, git_ref=, dockerfile=, context=, make_default=, wait_for_finish=)` · `upload_build(archive, version=, dockerfile=, context=)` · `build_of(id)` · `runs()` · `last_run(status=)` · `builds()` · `versions()`
`tasks()` / `task(id)`	`list` · `get` · `update` · `delete` · `start(task_input=, …)` · `call` · `get_input` · `update_input` · `runs()` · `last_run()`
`runs()` / `run(id)`	`list(status=, page=, per_page=)` · `get` · `wait_for_finish(wait_secs=None, poll_secs=2)` · `abort()` · `update(status_message=)` · `charge` · `resurrect(build=, memory_mb=, timeout_secs=)` · `reboot` · `metamorph(target_actor_id, run_input)` · `dataset()` · `key_value_store()` · `request_queue()` · `log().get()` / `log().read(offset)`
`datasets()` / `dataset(id or name)`	`list(unnamed=, offset=, limit=, desc=)` · `get_or_create(name=)` · `get` · `update(name=)` · `delete` · `list_items(offset=, limit=, clean=, desc=, fields=, omit=, unwind=, flatten=, skip_hidden=, view=)` → `ListPage` · `iterate_items(page_size=1000)` · `download_items(item_format="csv")` · `push_items`
`key_value_stores()` / `key_value_store(id)`	`list` · `get_or_create(name=)` · `list_keys` → `{items, is_truncated, next_exclusive_start_key}` · `iterate_keys` · `get_record(key)` → `{key, value, content_type}` · `get_record_as_bytes` · `record_exists` · `set_record` · `delete_record`
`request_queues()` / `request_queue(id)`	`add_request({url, unique_key?, user_data?…})` → `{request_id, was_already_present, was_already_handled}` · `batch_add_requests` · `list_requests(limit=, cursor=, handled=)` · `iterate_requests` · `delete_request`
`webhooks()` / `webhook(id)`	`list` · `create(event_types=, request_url=, actor_id=, task_id=, payload_template=, description=)` (the answer has the signing `secret`) · `get` · `update` · `delete` · `test` · `rotate_secret`
`schedules()` / `schedule(id)`	`list` · `create` · `get` (with `recent_runs`) · `update` · `delete`
`build(actor_id, build_id)`	`get` · `abort` · `wait_for_finish` · `log()`
`log(run_id)`	`get`
`me()`	`get`

Answers are the API's snake_case objects (default_dataset_id, status_message…). last_run(), run(id).dataset() and similar look up what they point at on first use.

Errors and retries

get() of something that does not exist returns None.
Other errors raise lantfly.ApiError with status, code, message, fields and details (see Errors).
Network errors, 429 and 5xx are retried with backoff, except calls that start or charge something (start, call, build, charge, metamorph…), which are sent once.

Async client

LantflyClientAsync has the same methods with await, and async for on the iterate_* methods.

async_client.py
import asyncio
from lantfly import LantflyClientAsync


async def main() -> None:
    client = LantflyClientAsync(base_url="https://lantfly.com")  # token from LANTFLY_TOKEN
    run = await client.actor("lantfly~hello").call(run_input={"count": 3})
    async for item in client.dataset(run["default_dataset_id"]).iterate_items():
        print(item)


asyncio.run(main())

Inside a run

Without arguments, the client reads LANTFLY_TOKEN and LANTFLY_API_BASE_URL. Inside a run, Actor.new_client() (or LantflyClient()) therefore uses the run token, which reaches only that run and its storages (and the account's named storages). See Actor environment.

Python actor template

A complete Python actor has main.py, requirements.txt, a Dockerfile and an .actor/ folder with actor.json, input_schema.json and dataset_schema.json. The quickstart shows the files (Python tab).

Exports

Actor, ActorClass, AbortingEvent, Dataset, KeyValueStore, RequestQueue, Log, ProxyConfiguration, ActorEnv, read_env, ApiBackend, LocalBackend, ApiError, LantflyClient, LantflyClientAsync, ListPage, TERMINAL_STATUSES, OutcomeTracker, TargetHandle, compute_status, status_message, ERROR_CODES, BLOCK_CODES, classify_block, PLATFORMS, review_item, review_rating, review_count, review_target_stats, ReviewItem, ReviewTarget, ReviewTargetStats, REVIEW_TARGET_TYPES, __version__.

Install​

Write an actor​

Try it locally​

Actor API​

Values​

Request queue​

Proxy​

Run outcome (OUTPUT)​

Graceful stop​

Review items​

Environment​

Call the API​

Constructor​

Resources​

Errors and retries​

Async client​

Inside a run​

Python actor template​

Exports​