Process background jobs in Python with Edgenivo Queues

When a customer clicks Place order, a checkout endpoint may save the order, email a receipt, and call the warehouse before returning a response. Slow email delivery keeps the customer waiting; a warehouse outage can make checkout appear to fail even after the order is saved.

Move those follow-up tasks into Edgenivo Queues so the endpoint waits for the queue to accept the order message. Python functions decorated with @subscribe process the message independently and retry failed deliveries within the configured limits.

Edgenivo manages the queue and runs the subscribers. Your application still needs to handle duplicate deliveries and track work that cannot be completed.

Copy link to headingOverview

In this guide, you'll learn how to:

• Explain how Edgenivo Queues delivers messages, and what at-least-once delivery means for your code
• Declare a typed topic and publish messages to it from Python
• Consume messages with @subscribe, and fan one message out to several independent consumers
• Control retries, delays, and duplicate sends
• Test the whole path locally, then deploy it with a FastAPI app on Edgenivo
• Choose between Queues, Workflows, and Cron Jobs for a given job

Copy link to headingWhat is Edgenivo Queues?

Edgenivo Queues stores messages in topics for independent consumer groups to process. Topics retain messages until they expire, even after a group acknowledges them. Each group tracks its own progress.

The Python SDK provides asynchronous and synchronous clients, Pydantic-backed topics, and an embedded queue for local tests. Import the async API from vercel.queue or the sync API from vercel.queue.sync.

The checkout route waits for send() to accept the message. Consumer groups then process it independently, with failed deliveries retried until the applicable attempt limit or message expiry.

Copy link to headingThe pieces you'll use

Copy link to headingHow a message moves

On Edgenivo, subscribers run in push mode. List their modules in pyproject.toml so Edgenivo can build private functions that receive queue deliveries. One generated function can handle several subscribers from the same module.

The SDK renews the message lease while the handler runs. Returning successfully acknowledges the delivery for that consumer group. If the handler raises, the message remains eligible for another attempt within its retention period and configured delivery limit.

Each consumer group gets its own copy of every message. Two subscribers in different groups both see every order. Two subscribers in the same group share the work, and each message goes to only one of them.

Copy link to headingAt least once, so make handlers safe to repeat

Queues uses at-least-once delivery, so a handler can receive the same message again after a timeout or crash. Messages must become available before their retention period expires to be delivered.

Use stable idempotency keys enforced by the email provider for receipts. For database writes, use an upsert or unique constraint.

Checking whether work already happened is only sufficient when the check and effect are protected against concurrent attempts and failures.

Copy link to headingPrerequisites

Before you begin, make sure you have:

• A Edgenivo account.
• Python 3.12 or a newer version supported by the Edgenivo Python runtime.
• Working knowledge of async/await, Pydantic, and FastAPI.
• An existing application with database, email, and warehouse integrations.

For local development, you also need uv, Node.js 24+, and the Edgenivo CLI.

Copy link to headingStep 1: Set up the project

Create a project, add FastAPI and the Edgenivo SDK, and link it:

uv init --python 3.12 checkout
cd checkout
uv add fastapi vercel
vercel link

The vercel package includes the Queues SDK and Pydantic. For a smaller install, use vercel-queue[typed] instead.

This example adds Queues to an existing checkout. Create app_clients.py to wrap your application's integrations before running the code. These adapters are application code, separate from the Edgenivo SDK:

All four methods are asynchronous. Keep provider credentials in server environment variables, and configure an alert or review process for records written by failed_jobs.

Create the following files, including an empty queues/__init__.py:

checkout/
├── pyproject.toml
├── uv.lock
├── app_clients.py    # adapters for your existing integrations
├── main.py           # FastAPI app: publishes messages (Step 3)
└── queues/
    ├── __init__.py
    └── orders.py     # topic and subscribers (Step 2)

Copy link to headingStep 2: Declare the topic and its subscribers

Define the shared order model, topic, and subscribers in queues/orders.py:

from pydantic import BaseModel
from vercel.queue import Topic, subscribe

from app_clients import email, warehouse


class Order(BaseModel):
    order_id: str
    email: str
    items: list[str]
    total_cents: int


orders = Topic[Order]("orders")


@subscribe(topic=orders, consumer_group="receipts")
async def send_receipt(order: Order) -> None:
    await email.send(
        to=order.email,
        template="receipt",
        data=order.model_dump(),
        idempotency_key=f"receipt-{order.order_id}",
    )


@subscribe(topic=orders, consumer_group="warehouse")
async def notify_warehouse(order: Order) -> None:
    await warehouse.create_shipment(
        order_id=order.order_id,
        items=order.items,
        idempotency_key=f"shipment-{order.order_id}",
    )

Both consumer groups receive each order. The SDK validates the payload as an Order before calling the handlers, and each adapter receives a stable idempotency key for its external operation.

These keys work only if the adapters enforce duplicate protection for the retry window you need. The Queues SDK does not deduplicate email or shipping calls. Explicit consumer group names also make these handlers easier to identify in the dashboard; omitted names are derived from the function's import path.

Now tell Edgenivo where the subscribers live. Add this to pyproject.toml:

[[tool.vercel.subscribers]]
entrypoint = "queues.orders"

The entrypoint is a module import path, not a file path: queues.orders, not queues/orders.py.

At build time, Edgenivo imports that module, finds every @subscribe in it, and turns them into a private function that the queue calls. You don't touch vercel.json.

Copy link to headingStep 3: Publish from your checkout route

Add the route below to main.py. It shows the queue handoff after saving an order. In a production checkout, authenticate the caller and construct the Order from records the caller is authorized to use. Derive prices, fulfillment items, and the recipient on the server.

from fastapi import FastAPI
from vercel.queue import send

from app_clients import db
from queues.orders import Order, orders

app = FastAPI()


@app.post("/api/orders")
async def place_order(order: Order):
    await db.orders.save(order)
    await send(orders, order, idempotency_key=f"order-{order.order_id}")
    return {"status": "accepted", "order_id": order.order_id}

The response waits for the database save and queue publish, but no longer waits for email or the warehouse. Importing the shared Order and topic keeps the sender and subscribers on the same message schema.

Saving and publishing are separate operations. If send() fails after the database save, retry with the same order and idempotency key.

To recover without client retries, save the order and an outbox record in one database transaction. Publish pending records from a separate process.

The idempotency_key deduplicates repeated publishes, including retries after a double-click or network timeout. Reusing the key during its window is accepted without delivering another copy. The window lasts for the shorter of 24 hours and the message's retention period.

Use the same order ID when retrying, and make the database save idempotent too. This key does not cover the database write or the subscribers' external side effects.

Copy link to headingStep 4: Handle failures and retries

Failed deliveries retry within the message's retention period, which defaults to 24 hours. The default retry delay is 60 seconds. For a warehouse outage, increase the delay between attempts and record unresolved orders for an operator to review.

Replace notify_warehouse in queues/orders.py with this version, keeping the Order, orders, and receipt handler from Step 2. The full Message[Order] gives the handler access to its delivery count:

from vercel.queue import Message, RetryAfter, subscribe

from app_clients import WarehouseUnavailable, failed_jobs, warehouse

MAX_SHIPMENT_ATTEMPTS = 8


@subscribe(topic=orders, consumer_group="warehouse")
async def notify_warehouse(message: Message[Order]) -> None:
    order = message.payload
    attempt = message.metadata.delivery_count

    try:
        await warehouse.create_shipment(
            order_id=order.order_id,
            items=order.items,
            idempotency_key=f"shipment-{order.order_id}",
        )
    except WarehouseUnavailable as exc:
        if attempt >= MAX_SHIPMENT_ATTEMPTS:
            await failed_jobs.record(
                key=f"warehouse-{order.order_id}",
                order=order.model_dump(),
                error=str(exc),
            )
            return
        raise RetryAfter(min(600, 30 * 2 ** (attempt - 1))) from exc

Requested delays double from 30 seconds to a maximum of 10 minutes. RetryAfter sets eligibility for redelivery; dispatch can happen later. Other exceptions use the subscriber's base retry_after setting.

After eight or more delivery attempts, a retryable warehouse error is recorded for operator review. The handler acknowledges only after that write succeeds. If the write fails, delivery can retry until the message expires. The stable failure-record key prevents duplicate review items.

Queues has no built-in dead-letter queue. max_attempts is a separate cap on push deliveries. It can stop a trigger from delivering a message even if your final error handler never completed, so this example leaves it unset. Monitor retry depth and the failed-job store independently; neither a delivery cap nor a notification call guarantees a successful handoff.

Handlers accept one argument, either a typed payload or a Message containing that payload and delivery metadata. Use the message form when you need its attempt count or message ID.

Copy link to headingStep 5: Run it locally

Start the app with the Edgenivo CLI:

vercel dev

With the application adapters in place, vercel dev runs FastAPI and an in-memory queue that dispatches to your subscribers. Use it to check the local integration; restarting it clears local queue state.

Place an order from a second terminal:

curl -X POST http://localhost:3000/api/orders \
  -H "Content-Type: application/json" \
  -d '{"order_id": "ord_1", "email": "[email protected]", "items": ["mug"], "total_cents": 1800}'

The endpoint returns an accepted status after saving the order and queuing its message. Use local logs or test adapters to confirm both subscribers ran.

To exercise retries, configure the warehouse adapter to raise WarehouseUnavailable in your local test environment and send a new order ID. The receipt handler can still succeed while the warehouse handler retries with increasing delays. Use test credentials or fake adapters for this exercise.

Run the app with vercel dev rather than plain uvicorn. Outside vercel dev, send() has no local queue to talk to, and it needs a region and a deployment ID before it will reach the real service.

Copy link to headingStep 6: Test your subscribers

The decorator returns the original function, so tests can call a handler directly. Install pytest for the test runner; AnyIO is included through the Queues SDK:

uv add --dev pytest

Create tests/conftest.py with these fakes. They replace the adapters imported by queues.orders and use the asyncio backend:

import asyncio
from types import SimpleNamespace

import pytest

import queues.orders as handlers


@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.fixture
def fake_email(monkeypatch):
    sent = []
    lock = asyncio.Lock()

    async def send(*, to, template, data, idempotency_key):
        async with lock:
            if idempotency_key not in sent:
                sent.append(idempotency_key)

    fake = SimpleNamespace(send=send, sent=sent)
    monkeypatch.setattr(handlers, "email", fake)
    return fake


@pytest.fixture
def fake_warehouse(monkeypatch):
    shipments = []
    keys = set()

    async def create_shipment(*, order_id, items, idempotency_key):
        if idempotency_key not in keys:
            keys.add(idempotency_key)
            shipments.append(order_id)

    fake = SimpleNamespace(create_shipment=create_shipment, shipments=shipments)
    monkeypatch.setattr(handlers, "warehouse", fake)
    return fake

Create tests/test_orders.py with a repeated-delivery test:

import asyncio

import pytest

from queues.orders import Order, send_receipt

ORDER = Order(order_id="ord_1", email="[email protected]", items=["mug"], total_cents=1800)


@pytest.mark.anyio
async def test_receipt_uses_a_stable_key(fake_email):
    await send_receipt(ORDER)
    await asyncio.gather(send_receipt(ORDER), send_receipt(ORDER))
    assert fake_email.sent == ["receipt-ord_1"]

This test checks key reuse with an idempotent fake. Verify the real email adapter's timeout retries and the provider's key retention period.

Use the embedded queue service to test publishing and dispatch together. It delivers messages to registered subscribers within the test process:

import anyio
from vercel.queue.embedded import embedded_queue_service

from queues.orders import orders  # importing the module registers its subscribers


@pytest.mark.anyio
async def test_order_reaches_both_consumers(fake_email, fake_warehouse):
    async with embedded_queue_service() as service:
        await service.get_async_client().send(orders, ORDER)
        with anyio.fail_after(5):
            while not (fake_email.sent and fake_warehouse.shipments):
                await anyio.sleep(0.01)

    assert fake_email.sent == ["receipt-ord_1"]
    assert fake_warehouse.shipments == ["ord_1"]

Append this test to tests/test_orders.py and run uv run pytest. It reuses ORDER and the fixtures above. Publishing through service.get_async_client() routes the message to the embedded service.

Wait for your handlers' side effects, not for the queue to report the message as handled. With two consumer groups on one topic, the embedded queue marks a message acknowledged as soon as either group finishes, so a test that waits on that can pass before the second handler has even run.

For tests that only send or poll messages, the SDK also provides an embedded_queue_server fixture. Enable it by adding pytest_plugins = ["vercel.queue.testing.pytest"] to tests/conftest.py. This fixture exposes the queue API without dispatching subscribers.

Copy link to headingStep 7: Deploy

Add .vercel to .gitignore if it isn't there, commit uv.lock, and deploy:

vercel deploy        # preview
vercel deploy --prod # production

Edgenivo detects FastAPI, reads [[tool.vercel.subscribers]], and builds your subscribers into a queue-triggered function. On Edgenivo, send() authenticates with the deployment's OIDC token automatically.

Push deliveries go to the publishing deployment by default, so orders queued before a release still run through that deployment's subscribers.

Old deployments can receive retries until messages are acknowledged, expire, or reach a delivery cap. Keep them available while work remains because promoting a release does not move the backlog to new code.

Open your project's Observability tab and select Queues. Each queue shows Max Message Age and Retry Depth by consumer group.

Rising Retry Depth means at least one message is being delivered repeatedly. Check the affected group's logs for handler errors or a downstream outage. With Observability Plus, the Queue Retry Depth alert preset can notify you when retries cross its threshold.

Copy link to headingUse cases

The following patterns build on the checkout example. The CRM, audit-log, and import snippets are integration excerpts: supply your own crm, audit_log, and loader clients before adding them to a subscriber module.

Copy link to headingDo something later

Set delay to postpone delivery and retention to leave time for processing afterward. This example delays a reminder for one day and retains it for two:

from datetime import timedelta

from pydantic import BaseModel
from vercel.queue import Topic, send


class CartReminder(BaseModel):
    cart_id: str


reminders = Topic[CartReminder]("cart-reminders")


async def schedule_reminder(cart_id: str) -> None:
    await send(
        reminders,
        CartReminder(cart_id=cart_id),
        delay=timedelta(hours=24),
        retention=timedelta(days=2),
        idempotency_key=f"cart-reminder-{cart_id}",
    )

Call schedule_reminder(cart_id) when a cart becomes eligible for a reminder. Its subscriber should check whether the purchase has since completed before sending email. Both delay and retention have a seven-day maximum; messages expire when retention ends, even if they have not been processed.

For one follow-up, a delayed message is the simplest tool. When a process has several waits and decisions in sequence, use a workflow instead (below).

Copy link to headingProtect a downstream API

If a subscriber calls an API with a rate limit, cap how many deliveries run at once:

@subscribe(topic=orders, consumer_group="crm-sync", max_concurrency=5)
async def sync_to_crm(order: Order) -> None:
    await crm.upsert_customer(email=order.email, last_order=order.order_id)

This caps concurrent push deliveries for this consumer, but account-wide CRM and requests-per-second limits need separate controls. Coordinate other consumers and follow the provider's retry guidance.

Copy link to headingHandle a family of events

Use a trailing * to subscribe to topics with a shared prefix, such as events written to an audit log:

@subscribe(topic="user-*")
async def audit_user_event(message: Message[dict[str, str]]) -> None:
    await audit_log.write(topic=message.metadata.topic, event=message.payload)

This handler receives events from user-created, user-deleted, and any other topic with the user- prefix.

Copy link to headingMove large payloads

Topics typed as bytes or str carry raw content, and streaming topics accept AsyncIterable[bytes]. The queue supports payloads up to 100 MB, but the subscriber still needs enough execution time and memory to process them.

For a CSV import, use an incremental loader that buffers incomplete records and handles text characters split across chunks:

from collections.abc import AsyncIterable

from vercel.queue import ByteStreamTransport, Topic, subscribe

imports = Topic[AsyncIterable[bytes]]("csv-imports", transport=ByteStreamTransport())


@subscribe(topic=imports)
async def load_rows(chunks: AsyncIterable[bytes]) -> None:
    async for chunk in chunks:
        await loader.feed(chunk)

The SDK uses ByteStreamTransport to read and write byte streams. Without it, an asynchronous iterable is treated as JSON and fails serialization.

Copy link to headingRun a polling worker on Edgenivo Daemons

If you need an always-on worker that pulls messages, run it on Edgenivo Daemons with poll mode. The SDK reads the OIDC token file provided to each instance.

Move the Step 4 handler to workers/warehouse.py and add an empty workers/__init__.py. Import Order from queues.orders, retain MAX_SHIPMENT_ATTEMPTS and the other imports, and remove @subscribe.

Delete the warehouse handler from queues/orders.py and leave the receipt subscriber. This avoids competing push and poll consumers.

Save the polling loop as worker.py:

import asyncio
import logging
import signal

from vercel.queue import ALL_DEPLOYMENTS, QueueClient

from queues.orders import orders
from workers.warehouse import notify_warehouse

logger = logging.getLogger(__name__)


async def main() -> None:
    queue = QueueClient(region="iad1", deployment=ALL_DEPLOYMENTS)
    stopping = asyncio.Event()
    asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, stopping.set)

    while not stopping.is_set():
        handled = False
        try:
            async for delivery in queue.poll(orders, "warehouse", limit=1):
                handled = True
                try:
                    async with delivery as message:
                        await notify_warehouse(message)
                except Exception:
                    logger.exception("Warehouse delivery failed; left unacknowledged")
        except Exception:
            logger.exception("Queue polling failed")

        if not handled and not stopping.is_set():
            try:
                await asyncio.wait_for(stopping.wait(), timeout=1)
            except TimeoutError:
                pass


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

The delivery context acknowledges a successful handler and handles RetryAfter directives. Other exceptions leave the message unacknowledged and propagate, so the loop logs them and continues. The application-level failure threshold from Step 4 still runs because it is part of the handler; decorator options such as max_attempts do not govern direct calls from this loop.

Copy link to headingSelect the deployment scope and region

Daemon instances do not receive VERCEL_DEPLOYMENT_ID. This worker uses ALL_DEPLOYMENTS so it can process retained orders across production rollouts. That scope also includes preview deployments, so use a separate topic for preview traffic. Set the client region to the region where the producer sends its messages.

Copy link to headingAllow time to finish a delivery

The loop requests one message at a time and checks for SIGTERM before the next poll. Edgenivo allows 30 seconds for shutdown by default. Configure terminationGracePeriodSeconds and provider request timeouts so the current delivery can finish within that window; otherwise it remains available for retry after its lease expires.

vercel dev does not run Daemons. Build and test the worker container separately before deploying it, using the Daemon guide for its service configuration.

Moving an existing Celery app? Edgenivo documents how to run Celery tasks on Edgenivo Queues.

Copy link to headingWhen to use something else

Copy link to headingNext steps

• Read the full Python SDK reference for transports, manual polling, leases, and every error class
• Learn how delivery, retention, and deployments interact in Queues concepts
• Watch throughput and retries with Queues observability
• Check pricing and limits before you send at volume