Python Worker

This guide shows you how to build a Priostack job worker in Python using the requests library.

Prerequisites

• Python 3.10 or later
• The requests library: pip install requests
• A Priostack API key set as PRIOSTACK_API_KEY environment variable

Complete Worker Implementation

#!/usr/bin/env python3
"""
Priostack Job Worker — Python implementation
"""
import logging
import os
import signal
import time
from datetime import datetime, timezone
from typing import Any

import requests

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger(__name__)

BASE_URL = "https://api.priostack.com"
WORKER_TYPE = "payment-processor"
MAX_JOBS = 5
POLL_TIMEOUT_MS = 30_000


class PriostackClient:
    def __init__(self, api_key: str, base_url: str = BASE_URL):
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": api_key,
            "Content-Type": "application/json",
        })
        self.base_url = base_url

    def activate_jobs(self) -> list[dict]:
        resp = self.session.post(
            f"{self.base_url}/api/v1/jobs/activate",
            json={
                "type": WORKER_TYPE,
                "maxJobsToActivate": MAX_JOBS,
                "requestTimeout": POLL_TIMEOUT_MS,
                "fetchVariables": ["orderId", "amount", "currency"],
            },
            timeout=35,
        )
        resp.raise_for_status()
        return resp.json().get("jobs", [])

    def complete_job(self, job_key: int, variables: dict[str, Any]) -> None:
        resp = self.session.post(
            f"{self.base_url}/api/v1/jobs/{job_key}/complete",
            json={"variables": variables},
            timeout=10,
        )
        resp.raise_for_status()

    def fail_job(self, job_key: int, retries: int, error_message: str) -> None:
        resp = self.session.post(
            f"{self.base_url}/api/v1/jobs/{job_key}/fail",
            json={
                "retries": max(0, retries - 1),
                "errorMessage": error_message,
                "retryBackoff": 5000,
            },
            timeout=10,
        )
        resp.raise_for_status()

    def throw_bpmn_error(self, job_key: int, error_code: str, message: str) -> None:
        resp = self.session.post(
            f"{self.base_url}/api/v1/jobs/{job_key}/error",
            json={"errorCode": error_code, "errorMessage": message},
            timeout=10,
        )
        resp.raise_for_status()


def process_payment(job: dict) -> dict[str, Any]:
    """Your actual business logic goes here."""
    variables = job.get("variables", {})
    order_id = variables.get("orderId", "unknown")
    amount = variables.get("amount", 0)

    log.info("Processing payment for order %s, amount %.2f", order_id, amount)

    # TODO: Call your actual payment gateway
    # Example: response = stripe.PaymentIntent.create(amount=int(amount*100), currency="eur")

    # Simulate processing time
    time.sleep(0.1)

    return {
        "transactionId": f"txn_{order_id}_{int(time.time())}",
        "paymentStatus": "success",
        "processedAt": datetime.now(timezone.utc).isoformat(),
    }


class Worker:
    def __init__(self, client: PriostackClient):
        self.client = client
        self.running = True
        signal.signal(signal.SIGTERM, self._shutdown)
        signal.signal(signal.SIGINT, self._shutdown)

    def _shutdown(self, signum, frame):
        log.info("Shutdown signal received. Stopping worker...")
        self.running = False

    def handle_job(self, job: dict) -> None:
        job_key = job["key"]
        retries = job.get("retries", 3)

        try:
            result = process_payment(job)
            self.client.complete_job(job_key, result)
            log.info("Job %d completed successfully", job_key)
        except Exception as exc:
            log.error("Job %d failed: %s", job_key, exc)
            try:
                self.client.fail_job(job_key, retries, str(exc))
            except Exception as fail_exc:
                log.error("Failed to report job failure: %s", fail_exc)

    def run(self) -> None:
        log.info("Worker started. Polling for jobs of type %r...", WORKER_TYPE)

        while self.running:
            try:
                jobs = self.client.activate_jobs()
                for job in jobs:
                    log.info(
                        "Activated job %d (instance %d)",
                        job["key"],
                        job.get("processInstanceKey", 0),
                    )
                    self.handle_job(job)
            except requests.exceptions.Timeout:
                # Long poll returned with no jobs — normal
                pass
            except requests.exceptions.HTTPError as exc:
                if exc.response is not None and exc.response.status_code == 429:
                    log.warning("Rate limited. Waiting 60s...")
                    time.sleep(60)
                else:
                    log.error("HTTP error: %s. Retrying in 5s...", exc)
                    time.sleep(5)
            except requests.exceptions.RequestException as exc:
                log.error("Network error: %s. Retrying in 5s...", exc)
                time.sleep(5)

        log.info("Worker stopped.")


def main():
    api_key = os.environ.get("PRIOSTACK_API_KEY")
    if not api_key:
        raise SystemExit("PRIOSTACK_API_KEY environment variable is required")

    client = PriostackClient(api_key)
    Worker(client).run()


if __name__ == "__main__":
    main()

Running the Worker

# Install dependencies
pip install requests

# Set your API key
export PRIOSTACK_API_KEY=ps_live_your_key_here

# Run the worker
python worker.py

# Output:
# 2026-05-01 14:00:00 INFO Worker started. Polling for jobs of type 'payment-processor'...
# 2026-05-01 14:00:05 INFO Activated job 2251799813685290 (instance 2251799813685281)
# 2026-05-01 14:00:05 INFO Processing payment for order ORD-001, amount 149.99
# 2026-05-01 14:00:05 INFO Job 2251799813685290 completed successfully

Dockerfile

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY worker.py .
CMD ["python", "worker.py"]

requirements.txt

requests==2.31.0

Production Tips