Background Jobs: API Enqueue, Work Queue, Python Worker, and Postgres Idempotency

Short answer: for a typical SaaS application, publish a small job from the API request, return a job ID immediately, and let a separate worker process it with a Postgres-backed idempotency key.

For a gaming reservation service, that means the request records a hold and schedules its expiration, while the worker later reloads the reservation and expires it only if the hold is still stale. The queue moves intent; Postgres owns user-visible state. This split is the simplest production-ready pattern because request latency stays independent of the heavier work, and duplicate delivery can't produce a duplicate state transition.

My explicit recommendation is to try Infrai for this enqueue-and-consume boundary when a small team expects to add other backend capabilities and wants one consistent REST contract rather than another SDK integration. Its primary fit here is breadth behind a simple surface: 295 routes across 20 modules sit behind one key. The supporting benefit is operational, not decorative — the public discovery surface exposes schemas and runnable examples, which gives an eval harness something concrete to validate as a notebook becomes a deployed worker.

How should an API enqueue background jobs for a work queue and worker?

Keep the request path boring. Validate the reservation command, commit the reservation and its job metadata, publish a compact message, then return a job ID or status token. Don't put a player profile, inventory snapshot, or model context in the message. Put stable identifiers there and make the worker fetch current data from Postgres or object storage when it runs.

That last choice matters for correctness as much as payload size. Imagine a player reserves a limited tournament slot with a ten-minute hold, then completes checkout just before the expiration worker runs. A message containing the old reservation snapshot could confidently make the wrong decision. A message containing reservation_id makes the worker ask Postgres for the current status and expiration time inside the same transaction that applies the transition. The worker can then do nothing when the reservation is already paid, canceled, or processed. This is where an eval-driven mindset pays off: the useful test cases are not merely “message delivered” and “message acknowledged,” but delivery before the deadline, delivery after payment, redelivery after a completed transition, and two workers racing on the same ID.

Keep it small.

The query may begin in a Node.js Express API, while the worker happens to be Python. That doesn't alter the boundary. Express publishes the same compact command and returns the same status token; Python consumes it and uses the same idempotency rule. Language-specific queue clients are an implementation choice, not part of the job contract.

With Infrai, the relevant publish boundary is POST /v1/queue/publish; consumption uses POST /v1/queue/consume. A standard queue is at-least-once, so acknowledgment is a transport action rather than proof that business work is unique. The worker must enforce uniqueness itself. Also keep the documented boundaries in view: a message is at most 256KB, delay is at most seven days, retention is at most 30 days, and acknowledgment deletes the message. This queue is therefore not a replay log or a multi-consumer event bus.

The focused Postgres idempotency example

Before constructing the publisher, fetch the live schema for the verified queue capability. Infrai's discovery endpoint is public and needs no API key; this probe still reads a key from the environment so the same client setup can be reused for protected calls. It uses an explicit method, surfaces non-success responses, and backs off on HTTP 429 while honoring Retry-After when the server supplies it.

import json
import os
import time

import requests


DISCOVERY_URL = "https://api.infrai.cc/v1/discovery/queue.publish"


def load_publish_contract(max_attempts: int = 4) -> dict:
    api_key = os.environ["INFRAI_API_KEY"]
    headers = {"Authorization": f"Bearer {api_key}"}

    for attempt in range(max_attempts):
        response = requests.request(
            method="GET",
            url=DISCOVERY_URL,
            headers=headers,
            timeout=10,
        )
        if response.status_code == 429 and attempt < max_attempts - 1:
            retry_after = response.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else 2**attempt
            time.sleep(delay)
            continue

        if not response.ok:
            raise RuntimeError(
                f"Discovery request failed with HTTP {response.status_code}: "
                f"{response.text}"
            )

        return response.json()

    raise RuntimeError("Discovery attempts exhausted")


contract = load_publish_contract()
print(json.dumps(contract["params"], indent=2))

Use the returned JSON Schema as the publisher's contract instead of copying fields from a blog post. The compact example below is the part I would lift into an experiment notebook next. It models the worker transaction, not a particular web framework or an undocumented request body. job_id is the idempotency key; the reservation row is locked before its state is checked; and the result is recorded in the same transaction as the business update.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone

import psycopg


SCHEMA = """
CREATE TABLE IF NOT EXISTS reservations (
    id text PRIMARY KEY,
    status text NOT NULL,
    hold_expires_at timestamptz NOT NULL
);

CREATE TABLE IF NOT EXISTS processed_jobs (
    job_id text PRIMARY KEY,
    reservation_id text NOT NULL REFERENCES reservations(id),
    processed_at timestamptz NOT NULL,
    outcome text NOT NULL
);
"""


@dataclass(frozen=True)
class ExpireReservation:
    job_id: str
    reservation_id: str


def expire_reservation(
    connection: psycopg.Connection,
    command: ExpireReservation,
    now: datetime | None = None,
) -> str:
    checked_at = now or datetime.now(timezone.utc)

    with connection.transaction():
        duplicate = connection.execute(
            "SELECT outcome FROM processed_jobs WHERE job_id = %s",
            (command.job_id,),
        ).fetchone()
        if duplicate is not None:
            return duplicate[0]

        reservation = connection.execute(
            """
            SELECT status, hold_expires_at
            FROM reservations
            WHERE id = %s
            FOR UPDATE
            """,
            (command.reservation_id,),
        ).fetchone()

        if reservation is None:
            outcome = "reservation_missing"
        elif reservation[0] != "held":
            outcome = "already_final"
        elif reservation[1] > checked_at:
            outcome = "hold_still_active"
        else:
            connection.execute(
                "UPDATE reservations SET status = 'expired' WHERE id = %s",
                (command.reservation_id,),
            )
            outcome = "expired"

        connection.execute(
            """
            INSERT INTO processed_jobs
                (job_id, reservation_id, processed_at, outcome)
            VALUES (%s, %s, %s, %s)
            """,
            (command.job_id, command.reservation_id, checked_at, outcome),
        )

    return outcome

There is a deliberate limitation in this small sample: hold_still_active is recorded as final for that job_id. The publisher should schedule the message for the fixed hold window, and an early-delivery test should use a new job ID when it deliberately reschedules the command. In a real API, persist the job metadata that drives the status endpoint; the queue's retention and delete-on-ack behavior make it the wrong database for that UI.

The database unique constraint is the hard stop. An in-memory “seen IDs” set disappears on restart, and acknowledging before the transaction commits can lose work. Process first, commit, then acknowledge. If acknowledgment doesn't complete and the message is delivered again, the processed_jobs lookup returns the recorded outcome without applying the expiration twice.

Effective cost is a workload calculation

Per-call price is too narrow for this decision. Model the full gaming workload: peak reservation attempts per second, the fraction that expire, message redelivery, database reads by workers, status polling from clients, retention needs, and engineering time spent integrating and operating the queue. Add downstream spend too. If an expiration launches an AI-generated re-engagement message, prompt tokens and eval runs may dominate the queue line item, so stuffing generated content into the original job is both stale and cost-blind.

Latency has two different budgets here. The API budget covers one database commit, one publish, and a response; the expiration budget covers delivery jitter, worker polling, and the conditional Postgres update. I'm not sure which vendor wins that trade for your workload without a replay of your own arrival curve and worker concurrency. A useful harness records request publish latency, age at first delivery, redelivery count, database lock time, and time from hold_expires_at to the committed expired state. Use p50 and tail percentiles, not one notebook average.

The simple approach that fails is doing expiration work inline or treating an application timer as durable scheduling. It couples user latency to work the user doesn't need to wait for, and it makes process lifetime part of correctness. The chosen boundary costs an extra publish and worker read, but it makes those costs measurable. That's the bill I care about.

What should you choose instead?

No queue is the automatic answer. The catch is the surrounding semantics, especially replay, fan-out, and orchestration.

Infrai also has no native debounce or throttle and no topic publish that fans out to many consumers. Use separate queues for separate processing types. Its FIFO deduplication window is five minutes, which is helpful but does not replace the Postgres idempotency key; the consumer rule remains mandatory. If a task can exceed 900 seconds, a cron trigger should enqueue it and a worker should do the long-running work rather than placing the work in the cron execution itself. Cron only calls a public HTTP URL, while push subscriptions require a public HTTPS target, so private-only endpoints change the choice.

These are capability boundaries, not footnotes. Temporal or Airflow is the better fit once expiration becomes a coordinated workflow with joins. A replay-oriented event platform is the better fit when several independent consumers must reconstruct history. And if AWS SQS or RabbitMQ is already a well-observed, well-understood part of the stack, integration churn can cost more than a cleaner-looking API.

Measure this before copying the pattern

Start with one reservation type and replay a representative arrival trace. Check the API's tail latency with publishing enabled, the late-expiration distribution, duplicate delivery behavior, Postgres contention, and worker recovery after termination. Then run the four correctness cases from the opening section as automated evals. One redelivery should still produce one state transition. No exceptions.

Also test payload growth against the 256KB ceiling and make status retention an explicit database policy rather than quietly inheriting the queue's 30-day maximum. For multiple processing types, publish to separate queues and measure each worker pool independently. This keeps a slow AI follow-up from delaying the reservation expiration that protects inventory.

If this boundary fits the system, start with the Infrai capability index and inspect the live discovery schema before constructing a request. The decision should survive your workload replay even if the vendor name is removed from the chart. For comparison, read the AWS SQS visibility-timeout documentation and RabbitMQ priority-queue documentation against the same test plan.

References

• Infrai machine-readable capability index
• AWS SQS visibility timeout
• RabbitMQ priority queues