Ideas for bq enhancements

[mahmoud]

Hey again! Per #3, I have a few ideas/observations after 3 months of using bq. For a bit of background, at SimpleLegal my team built a very similar transactional queuing system on postgres, but locked into Django land.

These days I can't go back to Django from SQLAlchemy and bq has been such a find, and fills an important gap in the Python queuing ecosystem.

Hoping to get your reactions before breaking ground on PRs. Happy to discuss about each:

1. More reliable metrics

In my production setup I've got monitoring hooked up to the metrics server as a healthcheck. It was such a pleasant surprise to find that already built. But after 3 months, I've observed inconsistencies in response times (5+ seconds) that require restarting the worker. Happens at least once every couple days. The worker itself is fine, but IME the Python builtins for http server stuff are pretty flaky. So, one idea is to improve this, whether via introducing a lightweight (pure python) dependency, or some other means.

2. Enhanced graceful shutdown

When running under docker compose and then exiting with an interrupt, all my other python processes terminate gracefully amost immediately, but I've noticed the bq worker wait for a full heartbeat sometimes. I suspect we could be doing some improved signal handling. Curious if you've seen this.

3. ULIDs

This is my main reason for using custom Task types (thank goodness bq supports them). I'm fine continuing to use my custom approach, but I think there's enough growing consensus around ULIDs as an ideal primary key that it might make sense to upstream a version for others to use? Happy to share what I'm doing if this is interesting.

4. Concurrency

Basically, this TODO already in the code.

This is probably the biggest place where I feel like it's hard to sell others on bq. Right now I'm running multiple instances of bq to scale, which is a little expensive, but manageable. There are so many ways to approach this, but here are the main thoughts I've had.

1. We'd like to keep it native (processes/threads/async) to avoid introducing heavy dependencies or unfamiliar concurrency paradigms.
2. Multiple processes has advantages but will at the very least require a better healthcheck, more like a supervisord-style architecture. In my past experience, we did this with forking and having a central process monitor the children and provide healthcheck/restarts.
3. When a fully loaded worker is 200MB+ as baseline (as is the case with my application), I start to wish I had threads/async for my lighter weight IO-bound tasks.
4. Async is appealing because sqlalchemy has async session support, and some applications only have the single async driver. In my application, I have both async and sync db drivers, and while the transactions can't be shared between them easily, I do use it on the service side to have some async endpoints and some sync endpoints.

---

Alright, that's it for now. Feels great to finally get that out there. We may not be able to fix it overnight, but hoping with a bit of discussion we can start small and go big! :)

[fangpenlin]

1. More reliable metrics

I haven't paid too much attentions into the metrics HTTP server's performance in production environment in the long run, but yeah, I can imagine that the built-in HTTP server is not so optimized for reliability in the long run. I agree that we should change it to something else that's more reliable provides if the built-in one indeed has problems like that. Of course, we don't want to introduce too much of dependencies. I haven't had a chance to study which one to use. But as I have been using FastAPI a lot recently, so maybe ASGI based stuff like these could help:

https://github.com/encode/starlette
https://www.uvicorn.org/

Open to suggestions. Or maybe open it up as a modular design? Hmmm 🤔

2. Enhanced graceful shutdown

Yeah, I also observed shutdown issue while developing locally. I recall the process somehow just hangs and I need to kill the process in order to restart. I have no time to look into the root cause yet as it doesn't cause me too much trouble at this moment. But sure, certainly there's a room for improvement.

3. ULIDs

I just checked it, first time heard this. Pretty cool stuff, I may use it in some of my other projects. Did you mean use ULIDs as the default ID column in BeanQueue? Because you can have custom task table, so I assume that's what you mean. And yeah, sure, I would love to learn how you use ULIDs.

4. Concurrency

100% agree this is something needs to be addressed before BeanQueue becomes a serious work queue project.

I think I will agree most of the improvement proposals, as myself I know there are plenty room for improvement. And I am thinking fixing them at some point. But unfortunately the only problem is that I have other big projects working on right now for my own startup. Unless it becomes a blocker there I won't be able to jump into an improvement here immediately. But yeah, we can start from smaller things first. PRs are welcomed. I, myself may also jump in and fix some of those above, just cannot guarantee when.

[mahmoud]

I know exactly what you mean re: priorities. Still, early alignment like this is a great starting point :)

Re: ULIDs, here's a copy of where most of my magic happens: https://gist.github.com/mahmoud/67cab15c68a59a97266bcbde53c4b9b4

3 notable customizations:

1. I'm using ULIDs, but specifically,
2. I'm using ObjIDs, which are my internal wrapper around TypeIDs: https://github.com/akhundMurad/typeid-python/ (these add a prefix to make IDs human-friendly and unambiguous. still a uuid at the db level. lmk if you want to see that implementation, too.)
3. I have a strong opinion on key naming convention, where it's always model_singular_id

So I have my own custom model for every primitive in bq. It was more work than I expected starting out, but compared to some other parts of my codebase they're all relatively straightforward.

[fangpenlin]

Thanks to #5, I guess now the "2. Enhanced graceful shutdown" part can be seen as done?

As for prefixed ULID or UUID7, interesting, I worked for a payment processing startup (Balanced Payments) long time ago they also had similar practice to have prefix + base58 of UUID or something like that (like CRDPYZEmNwuaRab84fuUj for credit cards ids). We basically had something almost identical like your ObjIDColumn(prefix='tptsk') in your sample code. At that time there was no ULID nor UUID7. I read a bit about it and even used ULID for one of my recent product release (the email id)

https://beanhub.io/blog/2025/04/21/new-beanhub-inbox-feature/

to make it sortable. I kind of like the idea. Overall I think it's a good idea. I happen to have a new project kick started recently. Usually I like to try news things out on a new project and bring the experience back to the others. I will try it out there and will probably bring what I've learned there back to bq.

[mahmoud]

Awesome! Agree that #5 handled #4.2. Thanks @Fazel94! :)

Sidnote, That's so wild you worked at Balanced. One of the founders @.mahmoudimus and I go way back to when his previous company (Milo) got acquired at eBay (where I was). I'm not surprised some of those Milo folks had some aligned ideas about the power of globally-unique primary keys!

Let me know how the experience goes. Been smooth sailing for me. Note that I have a tiny utility to translate between UUIDs and object IDs for different contexts (e.g., pgcli/sql output and REPL)

[fangpenlin]

Haha! Yeah small world 😄

So, I tried a bit on the typed id on my new project, here's what I do. Would love to know your feedbacks. I define a ModelID meant to be subclassed like this, along with ModelIDColumn class for the SQLAlchemy column.

import typing
import uuid

import sqlalchemy.types as types
from sqlalchemy import Dialect
from sqlalchemy.dialects.postgresql import UUID
from typeid import TypeID
from typeid.validation import validate_prefix


MODEL_PREFIX_ATTR = "PREFIX"


class PrefixValidator(type):
    def __new__(cls, name: str, bases: tuple[typing.Type], classdict: dict):
        # the "bases != (TypeID,)" part looks not necessary, but it's actually needed because
        # when ModelID class type created, this metaclass will be called, we don't want to
        # check model prefix for it
        if bases != (TypeID,) and bases == (ModelID,):
            if MODEL_PREFIX_ATTR not in classdict:
                raise ValueError(f"The {MODEL_PREFIX_ATTR!r} value is not defined")
            validate_prefix(classdict[MODEL_PREFIX_ATTR])
        return super().__new__(cls, name, bases, classdict)


class ModelID(TypeID, metaclass=PrefixValidator):
    PREFIX: str

    def __init__(self, prefix: str | None = None, suffix: str | None = None) -> None:
        if not hasattr(self, MODEL_PREFIX_ATTR):
            raise ValueError(
                "This class is not meant to be used directly but inherited"
            )
        if prefix is None:
            prefix = self.PREFIX
        elif prefix != self.PREFIX:
            raise ValueError(
                f"Expect prefix {self.PREFIX!r} but got {prefix!r} instead"
            )
        super().__init__(prefix=prefix, suffix=suffix)


class ModelIDColumn(types.TypeDecorator):
    impl = UUID
    cache_ok = True

    def __init__(self, model_id_cls: typing.Type[ModelID]):
        super().__init__(as_uuid=True)
        self.model_id_cls = model_id_cls

    def process_bind_param(self, value: TypeID, dialect: Dialect) -> str | None:
        if value is None:
            return None
        if not isinstance(value, self.model_id_cls):
            raise ValueError(
                f"Expected model id type {self.model_id_cls.__name__} but got {value.__class__.__name__} instead"
            )
        if value.prefix != self.model_id_cls.PREFIX:
            raise ValueError(
                f"Expected prefix as {self.model_id_cls.PREFIX} but got {value} instead"
            )
        return str(value.uuid)

    def process_result_value(
        self, value: str | TypeID | uuid.UUID | None, dialect: Dialect
    ) -> TypeID | None:
        if value is None:
            return None
        if isinstance(value, self.model_id_cls):
            return value
        return self.model_id_cls.from_uuid(value)

With that, you can define different model id types like:

from .types import ModelID


class ReportID(ModelID):
    PREFIX = "report"


class VerdictID(ModelID):
    PREFIX = "verdict"

Then, in your model definition, you use it directly like this:

import datetime

from sqlalchemy import DateTime
from sqlalchemy import String
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

from ..db.base import Base
from .helpers import make_repr_attrs
from .id_types import ReportID
from .types import ModelIDColumn


class Report(Base):
    id: Mapped[ReportID] = mapped_column(
        ModelIDColumn(model_id_cls=ReportID),
        primary_key=True,
        default=ReportID,
    )
    name: Mapped[str] = mapped_column(String, nullable=False)

    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime, nullable=False, default=datetime.datetime.utcnow
    )

    def __repr__(self) -> str:
        items = [
            ("id", self.id),
            ("name", self.name),
        ]
        return f"<{self.__class__.__name__} {make_repr_attrs(items)}>"

If you pass a wrong type to the id value, the type check will first complain. Then during the runtime, the prefix check will fail. In that way, we have an actual typed prefixed id system. The default function can also be passed in type obj ReportID directly.

[Fazel94]

Hello there,

I can contribute points 4.4 and 3 of the issue.
Could you give your opinions on them?
@mahmoud @fangpenlin