DBOS Transact

DBOS queues allow you to run tasks (single steps or entire workflows) in the background using Postgres as the backend. DBOS guarantees task completion and result retrieval even if the application is interrupted.

Key capabilities include:

• Flow control (limiting concurrency per queue or process).
• Task timeouts and rate limiting.
• Task deduplication and prioritization.

To use a queue, instantiate a Queue object and use queue.enqueue(function, args) to add tasks. You can then use the returned handle to retrieve results via handle.get_result().

from dbos import DBOS, Queue

queue = Queue("example_queue")

@DBOS.step()
def process_task(task):
  ...

@DBOS.workflow()
def process_tasks(tasks):
  task_handles = []
  # Enqueue each task so all tasks are processed concurrently.
  for task in tasks:
    handle = queue.enqueue(process_task, task)
    task_handles.append(handle)
  # Wait for each task to complete and retrieve its result.
  # Return the results of all tasks.
  return [handle.get_result() for handle in task_handles]

DBOS is a lightweight, Postgres-backed library for durable execution. Use the following comparisons to decide if DBOS fits your requirements:

DBOS vs. Temporal

• DBOS: Best if you want to add durable workflows with minimal rearchitecting and already use Postgres. It is implemented as a library where you annotate workflows and steps.
• Temporal: Best if you do not want to add Postgres to your stack or require a language not yet supported by DBOS. Temporal requires an externally orchestrated server and moving workflows/activities to a Temporal worker.

DBOS vs. Airflow

• DBOS: Best for general-purpose workflows written as code, especially when you need higher performance for streaming or real-time use cases. It requires only Postgres.
• Airflow: Best if you need Airflow's specific ecosystem of out-of-the-box connectors. Airflow is designed for batch operations and requires workflows to be written as explicit DAGs orchestrated from an Airflow cluster.

DBOS vs. Celery/BullMQ

• DBOS: Best when you need the reliability of enqueueing tasks from durable workflows. DBOS queues are durable and Postgres-backed, meaning workflows and tasks are checkpointed to guarantee completion even during failures.
• Celery/BullMQ: Best if you do not need durability or require extremely high throughput that exceeds what your Postgres server can handle. These are typically Redis-backed and do not provide workflow abstractions.

DBOS enables reliable event processing (e.g., webhooks, Kafka consumers) by ensuring workflows start exactly once in response to an event.

For webhooks, use SetWorkflowID(event_id) as a context manager to provide an idempotency key. For Kafka, use the @DBOS.kafka_consumer(config, [topics]) decorator on a @DBOS.workflow() function to ensure each message triggers a workflow exactly once.

# Webhook example
def handle_message(request: Request) -> None:
  event_id = request.body["event_id"]
  # Use the event ID as an idempotency key to start the workflow exactly-once
  with SetWorkflowID(event_id):
    # Start the workflow in the background, then acknowledge the event
    DBOS.start_workflow(message_workflow, request.body["event"])

# Kafka example
@DBOS.kafka_consumer(config,["alerts-topic"])
@DBOS.workflow()
def process_kafka_alerts(msg):
    # This workflow runs exactly-once for each message sent to the topic
    alerts = msg.value.decode()
    for alert in alerts:
        respond_to_alert(alert)