BeanQueue

BeanQueue, a lightweight Python task queue framework based on SQLAlchemy 2.0 asyncio, psycopg3, PostgreSQL SKIP LOCKED queries and NOTIFY / LISTEN statements.

Notice: Still in its early stage, we built this for BeanHub's internal usage. May change rapidly. Use at your own risk for now.

Features

• Super lightweight: Under 1K lines
• Easy-to-deploy: Only relies on PostgreSQL
• Transactional: Commit your tasks with other database entries altogether without worrying about data inconsistencies
• Easy-to-use: Built-in command line tools for processing tasks and helpers for generating task models
• Auto-notify: Automatic generation of NOTIFY statements for new or updated tasks, ensuring fast task processing
• Retry: Built-in and customizable retry policies
• Schedule: Schedule tasks to run later
• Worker heartbeat and auto-reschedule: Each worker keeps updating heartbeat, if one is found dead, the others will reschedule the tasks
• Custom health checks: Optional HTTP /healthz endpoint with pluggable checks via Blinker signals
• Customizable: Custom Task, Worker and Event models. Use it as a library and build your own work queue

Install

pip install beanqueue

To enable the optional metrics HTTP server (currently /healthz only), install with the metrics extra:

pip install "beanqueue[metrics]"

Testing

Unit and in-process acceptance tests:

uv run python -m pytest tests

End-to-end tests start PostgreSQL and three worker containers with Docker Compose, enqueue real tasks, apply load, kill a worker, and check graceful shutdown cleanup:

uv run python -m pytest tests/e2e -svvvv

Upgrading to 2.0

BeanQueue 2.0 is asyncio-first (SQLAlchemy AsyncSession + psycopg3) and includes breaking changes:

• Workers use asyncio SQLAlchemy.BeanQueue.make_session() returns an AsyncSession, engine is an AsyncEngine, and process_tasks() is async. Run it with asyncio.run(app.process_tasks(...)) or await it from your own loop.
• Database URLs use the psycopg3 driver. Prefer postgresql+psycopg://.... Bare postgresql://... URLs are normalized to postgresql+psycopg://....
• MAX_WORKER_THREADS is now MAX_CONCURRENT_TASKS.BQ_MAX_WORKER_THREADS and the old keyword still work as a deprecated alias.
• Processors may be async def or def. Async processors receive db: AsyncSession. Sync processors that take db still receive a sync Session via AsyncSession.run_sync().
• METRICS_HTTP_SERVER_ENABLED defaults to False (it was True in 1.x). Set BQ_METRICS_HTTP_SERVER_ENABLED=true to turn the server back on.
• The metrics server requires optional dependencies. Install beanqueue[metrics] (starlette and uvicorn). Without them, enabling the server raises an error at startup.
• Custom health checks use the healthz_check event (bq.events.healthz_check) instead of a healthz_check argument on bq.BeanQueue. Receivers that query the database should use AsyncSession (await session.execute(...)).

Usage

You can define a basic task processor like this

fromsqlalchemyimportselectfromsqlalchemy.ext.asyncioimportAsyncSessionimportbqfrom .. importmodelsfrom .. importimage_utilsapp=bq.BeanQueue()
@app.processor(channel="images")asyncdefresize_image(db: AsyncSession, task: bq.Task, width: int, height: int):
image= (
awaitdb.execute(select(models.Image).where(models.Image.task==task))
).scalar_one()
image_utils.resize(image, size=(width, height))
db.add(image)
# by default the `processor` decorator has `auto_complete` flag turns on,# so it will commit the db changes for us automatically

Synchronous processors are still supported. If the function takes a db argument, BeanQueue injects the sync Session behind the AsyncSession. Processors that do not use the database can stay as plain def functions and are run in a worker thread so they do not block the event loop.

The db and task keyword arguments are optional. If you don't need to access the task object, you can simply define the function without these two parameters. We also provide an optional savepoint argument in case if you want to rollback database changes you made.

To submit a task, you can either use bq.Task model object to construct the task object, insert into the database session and commit.

importbqfrom .dbimportSessionfrom .. importmodelsdb=Session()
task=bq.Task(
channel="files",
module="my_pkgs.files.processors",
name="upload_to_s3_for_backup",
)
file=models.File(
task=task,
blob_name="...",
)
db.add(task)
db.add(file)
db.commit()

Or, you can use the run helper like this:

from .processorsimportresize_imagefrom .dbimportSessionfrom .. importmy_modelsdb=Session()
# a Task model generated for invoking resize_image functiontask=resize_image.run(width=200, height=300)
# associate task with your own modelsimage=my_models.Image(task=task, blob_name="...")
db.add(image)
# we have Task model SQLALchemy event handler to send NOTIFY "<channel>" statement for you,# so that the workers will be woken up immediatelydb.add(task)
# commit will make the task visible to worker immediatelydb.commit()

To run the worker, you can do this:

BQ_PROCESSOR_PACKAGES='["my_pkgs.processors"]' bq process images

The BQ_PROCESSOR_PACKAGES is a JSON list contains the Python packages where you define your processors (the functions you decorated with bq.processors.registry.processor). To submit a task for testing purpose, you can do

bq submit images my_pkgs.processors resize_image -k '{"width": 200, "height": 300}'

To create tables for BeanQueue, you can run

bq create_tables

Schedule

In most cases, a task will be executed as soon as possible after it is created. To run a task later, you can set a datetime value to the scheduled_at attribute of the task model. For example:

importdatetimedb=Session()
task=resize_image.run(width=200, height=300)
task.scheduled_at=func.now() +datetime.timedelta(minutes=3)
db.add(task)

Please note that currently, workers won't wake up at the next exact moment when the scheduled tasks are ready to run. It has to wait until the polling times out, and eventually, it will see the task's scheduled_at time exceeds the current datetime. Therefore, depending on your POLL_TIMEOUT setting and the number of your workers when they started processing, the actual execution may be inaccurate. If you set the POLL_TIMEOUT to 60 seconds, please expect less than 60 seconds of delay.

Retry

To automatically retry a task after failure, you can specify a retry policy to the processor.

importdatetimeimportbqfromsqlalchemy.ext.asyncioimportAsyncSessionapp=bq.BeanQueue()
delay_retry=bq.DelayRetry(delay=datetime.timedelta(seconds=120))
@app.processor(channel="images", retry_policy=delay_retry)asyncdefresize_image(db: AsyncSession, task: bq.Task, width: int, height: int):
# resize image here ...pass

Currently, we provide some simple common retry policies such as DelayRetry and ExponentialBackoffRetry. You can define your retry policy easily by making a function that returns an optional object at the next scheduled time for retry.

defmy_retry_policy(task: bq.Task) ->typing.Any:
# Calculate delay based on task model ...returnfunc.now() +datetime.timedelta(seconds=delay)

To cap how many attempts are allowed, you can also use LimitAttempt like this:

delay_retry=bq.DelayRetry(delay=datetime.timedelta(seconds=120))
capped_delay_retry=bq.LimitAttempt(3, delay_retry)
@app.processor(channel="images", retry_policy=capped_delay_retry)asyncdefresize_image(db: AsyncSession, task: bq.Task, width: int, height: int):
# Resize image here ...pass

You can also retry only for specific exception classes with the retry_exceptions argument.

@app.processor(channel="images",retry_policy=delay_retry,retry_exceptions=ValueError,)asyncdefresize_image(db: AsyncSession, task: bq.Task, width: int, height: int):
# resize image here ...pass

Configurations

Configurations can be modified by setting environment variables with BQ_ prefix. For example, to set the python packages to scan for processors, you can set BQ_PROCESSOR_PACKAGES. To change the PostgreSQL database to connect to, you can set BQ_DATABASE_URL. The complete definition of configurations can be found at the bq/config.py module.

If you want to configure BeanQueue programmatically, you can pass in Config object to the bq.BeanQueue object when creating. For example:

importbqfrom .my_configimportconfigconfig=bq.Config(
PROCESSOR_PACKAGES=["my_pkgs.processors"],
DATABASE_URL=config.DATABASE_URL,
BATCH_SIZE=10,
)
app=bq.BeanQueue(config=config)

Then you can pass --app argument (or -a for short) pointing to the app object to the process command like this:

bq -a my_pkgs.bq.app process images

Or if you prefer to define your own process command, you can also call process_tasks of the BeanQueue object directly like this:

importasyncioasyncio.run(app.process_tasks(channels=("images",)))

Health check and metrics HTTP server

When enabled, each worker starts a small HTTP server (Starlette + Uvicorn) for operational endpoints. Today this only exposes GET /healthz, which returns {"status": "ok"} by default.

Enable it with the metrics extra installed and configuration:

pip install "beanqueue[metrics]"
BQ_METRICS_HTTP_SERVER_ENABLED=true bq process images

Relevant settings (see bq/config.py):

Setting	Env var	Default
METRICS_HTTP_SERVER_ENABLED	BQ_METRICS_HTTP_SERVER_ENABLED	False
METRICS_HTTP_SERVER_INTERFACE	BQ_METRICS_HTTP_SERVER_INTERFACE	"" (all interfaces)
METRICS_HTTP_SERVER_PORT	BQ_METRICS_HTTP_SERVER_PORT	8000
METRICS_HTTP_SERVER_LOG_LEVEL	BQ_METRICS_HTTP_SERVER_LOG_LEVEL	30 (WARNING)

Access requests are logged at INFO via uvicorn.access (visible even when METRICS_HTTP_SERVER_LOG_LEVEL is WARNING). BeanQueue also uses a metrics_server logger for its own messages. Override the entire logging setup by passing a logging.config dict via METRICS_HTTP_SERVER_LOG_CONFIG (or BQ_METRICS_HTTP_SERVER_LOG_CONFIG as JSON):

importbqconfig=bq.Config(
METRICS_HTTP_SERVER_ENABLED=True,
METRICS_HTTP_SERVER_LOG_CONFIG={
"version": 1,
"disable_existing_loggers": False,
"handlers": {
"default": {
"class": "logging.StreamHandler",
"formatter": "default",
}
},
"formatters": {
"default": {
"format": "%(asctime)s %(name)s %(levelname)s %(message)s",
}
},
"loggers": {
"metrics_server": {"handlers": ["default"], "level": "INFO"},
"uvicorn.access": {"handlers": ["default"], "level": "INFO"},
},
},
)
app=bq.BeanQueue(config=config)

Custom health checks

Register additional checks by connecting receivers to bq.events.healthz_check. If no receivers are connected, /healthz returns OK without touching the database.

With receivers connected, BeanQueue loads the current worker and passes a database session to each check. Receivers may be synchronous or asynchronous; both can be mixed on the same signal.

fromsqlalchemyimporttextfromsqlalchemy.ext.asyncioimportAsyncSessionimportbqfrombqimporteventsapp=bq.BeanQueue()
@events.healthz_check.connectasyncdefcheck_database(sender: bq.BeanQueue, worker, session: AsyncSession):
awaitsession.execute(text("SELECT 1"))
@events.healthz_check.connectasyncdefcheck_external_service(sender: bq.BeanQueue, worker, session: AsyncSession):
# async HTTP call, etc.
...

Receiver signature must accept the keyword arguments you need, for example (sender, worker, session), or use (sender, **kwargs). If a check raises an exception, /healthz responds with HTTP 500 and a JSON body containing the error message.

Define your own tables

BeanQueue is designed to be as customizable as much as possible. One of its key features is that you can define your own SQLAlchemy model instead of using the ones we provided.

To make defining your own Task, Worker or Event model much easier, use bq's mixin classes:

• bq.TaskModelMixin: provides task model columns
• bq.TaskModelRefWorkerMixin: provides foreign key column and relationship to bq.Worker
• bq.TaskModelRefParentMixin: provides foreign key column and relationship to children bq.Task created during processing
• bq.TaskModelRefEventMixin: provides foreign key column and relationship to bq.Event
• bq.WorkerModelMixin: provides worker model columns
• bq.WorkerRefMixin: provides relationship to bq.Task
• bq.EventModelMixin: provides event model columns
• bq.EventModelRefTaskMixin: provides foreign key column and relationship to bq.Task

Here's an example for defining your own Task model:

importuuidfromsqlalchemyimportForeignKeyfromsqlalchemy.dialects.postgresqlimportUUIDfromsqlalchemy.ormimportMappedfromsqlalchemy.ormimportmapped_columnfromsqlalchemy.ormimportrelationshipimportbqfrombq.models.taskimportlisten_eventsfrom .base_classimportBaseclassTask(bq.TaskModelMixin, Base):
__tablename__="task"worker_id: Mapped[uuid.UUID] =mapped_column(
UUID(as_uuid=True),
ForeignKey("worker.id", onupdate="CASCADE"),
nullable=True,
index=True,
)
worker: Mapped["Worker"] =relationship(
"Worker", back_populates="tasks", uselist=False
)
listen_events(Task)

For task insertion and updates to notify workers, we need to register any custom task types with bq.models.task.listen_events. In the example above, this is done right after the Task model definition. For more details and advanced usage, see the definition of bq.models.task.listen_events.

You just see how easy it is to define your Task model. Now, here's an example for defining your own Worker model:

importbqfromsqlalchemy.ormimportMappedfromsqlalchemy.ormimportrelationshipfrom .base_classimportBaseclassWorker(bq.WorkerModelMixin, Base):
__tablename__="worker"tasks: Mapped[list["Task"]] =relationship(
"Task",
back_populates="worker",
cascade="all,delete",
order_by="Task.created_at",
)

With the model class ready, you only need to change the TASK_MODEL, WORKER_MODEL and EVENT_MODEL of Config to the full Python module name plus the class name like this.

importbqconfig=bq.Config(
TASK_MODEL="my_pkgs.models.Task",
WORKER_MODEL="my_pkgs.models.Worker",
EVENT_MODEL="my_pkgs.models.Event",
# ... other configs
)
app=bq.BeanQueue(config)

Why?

There are countless work queue projects. Why make yet another one? The primary issue with most work queue tools is their reliance on a standalone broker server. Our work queue tasks frequently interact with the database, and the atomic nature of database transactions is great for data integrity. However, integrating an external work queue into the system presents a risk. The work queue and the database don't share the same data view, potentially compromising data integrity and reliability.

For example, you have a table of images to keep the user-uploaded images. And you have a background work queue for resizing the uploaded images into different thumbnail sizes. So, you will first need to insert a row for the uploaded image about the job into the database before you push the task to the work queue.

Say you push the task to the work queue immediately after you insert the images table then commit like this:

1. Insert into the "images" table
2. Push resizing task to the work queue
3. Commit db changes

While this might seem like the right way to do it, there's a hidden bug. If the worker starts too fast before the transaction commits at step 3, it will not be able to see the new row in images as it has not been committed yet. One may need to make the task retry a few times to ensure that even if the first attempt failed, it could see the image row in the following attempt. But this adds complexity to the system and also increases the latency if the first attempt fails. Also, if the commit step fails, you will have a failed work queue job trying to fetch a row from the database that will never exist.

Another approach is to push the resize task after the database changes are committed. It works like this:

1. Insert into the "images" table
2. Commit db changes
3. Push resizing task to the work queue

With this approach, we don't need to worry about workers picking up the task too early. However, there's another drawback. If step 3 for pushing a new task to the work queue fails, the newly inserted images row will never be processed. There are many solutions to this problem, but these are all caused by inconsistent data views between the database and the work queue storage. Things would be much easier if we had a work queue that shared the same consistent view as the database.

By using a database as the data storage, all the problems are gone. You can simply do the following:

1. Insert into the "images" table
2. Insert the image resizing task into the `tasks` table
3. Commit db changes

It's all or nothing! By doing so, you don't need to maintain another work queue backend. You are probably using a database anyway, so this work queue comes for free.

Usually, a database is inefficient as the work queues data storage because of the potential lock contention and the need for constant querying. However, things have changed since the introduction of the SKIP LOCKED and LISTEN / NOTIFY features in PostgreSQL or other databases.

This project is inspired by many of the SKIP-LOCKED-based work queue successors. Why don't we just use those existing tools? Well, because while they work great as work queue solutions, they don't take advantage of writing tasks and their relative data into the database in a transaction. Many provide an abstraction function or gRPC method for pushing tasks into the database, rather than allowing users to directly insert rows and commit them together.

BeanQueue doesn't overly abstract the logic of publishing a new task into the queue. Instead, you insert rows directly, choosing when and what to commit as tasks.

Sponsor

A modern accounting book service based on the most popular open source version control system Git and text-based double entry accounting book software Beancount.

Alternatives

• solid_queue
• good_job
• graphile-worker
• postgres-tq
• pq
• PgQueuer
• hatchet
• procrastinate