seekapi 1.0.2

SDK for SeekAPI workers: input/output handling, push data and files, success/failure.

SeekAPI Python SDK

SDK for building SeekAPI workers in Python. Handles input (fetch from presigned URL) and output (push JSON and files), with a minimal Lambda return contract.

• Zero dependencies — uses only the standard library.
• Python 3.10+ — works in AWS Lambda runtimes.

Install

pip install seekapi

Quick start (recommended)

Use run() so the SDK handles input, output, and errors. You only implement a function from input → output:

from seekapi import run

def handler(event, context):
    return run(event, context, lambda input: {
        "message": f"hello {input.get('name', 'world')}"
    })

Low-level API

When you need more control (e.g. multiple push_data calls or push_file):

from seekapi import (
    get_input,
    create_context,
    push_data,
    push_file,
    register_request_id,
    success,
    failure,
)

def handler(event, context):
    request_id = getattr(context, "aws_request_id", None) if context else None
    job_uuid = (event.get("job_uuid") or "").strip()
    if request_id and job_uuid:
        register_request_id(job_uuid, request_id)

    ctx = create_context(event)
    try:
        input_data = get_input(event)   # fetches and parses JSON from input_presigned_url
        # ... your logic ...
        push_data(ctx, {"result": "ok"})
        return success(request_id=request_id)
    except Exception as e:
        return failure("WORKER_ERROR", str(e), request_id=request_id)

API reference

Function	Description
get_input(event, timeout=10)	Fetch and parse job input JSON from event["input_presigned_url"]. Raises MissingInputError if URL missing, ValueError on HTTP/JSON errors.
create_context(event)	Build context dict for push_data/push_file + budget info (max_job_billing_usd, billing_mode, unit_price_usd).
push_data(context, data, type="json", bill=False, billing_push_id=None, timeout=8)	Push JSON to the job; set bill=True to debit one unit in per_unit.
push_file(context, name, local_path, content_type=None, bill=False, billing_push_id=None, timeout=15)	Push file to temp_files; set bill=True to debit one unit in per_unit.
register_request_id(job_uuid, request_id, timeout=3)	Register Lambda request_id for live logs (no-op if env not set).
success(request_id=None)	Return { "ok": true, "request_id"? }.
failure(code, message, request_id=None)	Return { "ok": false, "error": { "code", "message" }, "request_id"? }.
run(event, context, user_fn)	Get input → call user_fn(input) → push result → return success; on exception return failure. run() adapte les timeouts réseau au temps restant Lambda pour éviter les timeouts Sandbox.Timedout.

Environment (set by the platform)

• WORKER_API_BASE_URL — backend base URL for push and register_request_id.
• WORKER_INTERNAL_SECRET — secret for internal API auth.

Errors

• MissingInputError — event has no input_presigned_url.
• ValueError — Input fetch failed (HTTP error, timeout, invalid JSON). Message is descriptive.

Publishing to PyPI

From the package directory:

pip install build twine
python -m build
twine upload dist/*

Use a version bump and tag before each release.