queuebridge

Bidirectional Pydantic serialization for Celery, Dramatiq, and Arq.

Pass Pydantic models to .delay(), .send(), or enqueue_job(). Get models back from results. One shared wire codec powers every backend.

---

New here? Start with Getting started, then pick your backend tutorial:

• Celery tutorial

• Dramatiq tutorial

• Arq tutorial

---

Contents

Start here

• Getting started
  • What is queuebridge?
  • What you need to know
  • The problem in 30 seconds
  • Minimal example (Celery)
  • Next steps
• Installation
  • Requirements
  • From PyPI
  • With a task queue backend
  • From source (development)
  • Verify the install
  • Optional: build the docs locally

Tutorials (by backend)

• Celery tutorial
  • Step 1: Install
  • Step 2: Define your models
  • Step 3: Create the Celery app and register queuebridge
  • Step 4: Write your task
  • Step 5: Enqueue with a model
  • Step 6: Get a typed result
  • Full example
  • Testing without Redis
  • Common mistakes
  • API reference
• Dramatiq tutorial
  • Step 1: Install
  • Step 2: Register before defining actors
  • Step 3: Use validate_call on your actor
  • Step 4: Send a model
  • How it works internally
  • Full example
  • API reference
• Arq tutorial
  • Step 1: Install
  • Step 2: Get the serializer pair
  • Step 3: Configure worker and client
  • Step 4: Decorate your task with qb_task
  • Step 5: Typed job results
  • Full example
  • API reference

Concepts

• How it works
  • The flow (Celery example)
  • What each backend does
  • Type hints matter
  • When you do not need tags
• Wire format
  • The __qb__ envelope
  • Constants (Python)
  • Types and their wire representation
  • Encode and decode API
  • Celery note
  • Arq note
• Why not Celery pydantic=True alone?
  • What pydantic=True does
  • What it does not do
  • Official docs quote
  • Community reports
  • What queuebridge adds

Reference

• API reference
  • Core
  • Codec
  • Types and exceptions
  • Hints
  • Celery
  • Dramatiq
  • Arq
• Cookbook
  • Nested models and collections
  • Optional and Union fields
  • Testing Celery without a broker
  • Testing Dramatiq with StubBroker
  • Using encode/decode outside task queues
  • Manual wire inspection
  • Shared models across services
  • FastAPI + Celery pattern
  • Arq client + worker checklist
  • Common errors
  • Smoke test script
• Security
  • Trusted brokers only
  • What gets imported
  • Mitigations today
  • Planned for v0.2
  • Pickle vs msgpack (Arq)
  • Reporting issues
• FAQ
  • Can I use Pydantic v1?
  • Do I still need pydantic=True on Celery tasks?
  • Can I pass model_dump() dicts instead of models?
  • Why does ar.get() return a dict?
  • Will you monkey-patch AsyncResult.get()?
  • Does queuebridge work with shared_task?
  • What about task chords and chains?
  • Can I use pickle?
  • How do I test without Redis?
  • How do I publish a new version?
  • Where is the documentation hosted?

Indices and tables

• Index

• Module Index

• Search Page