Skip to content

Changelog

Version 0.3.0

Added

  • alchemical_queues.aio: an async counterpart to the sync API (AsyncAlchemicalQueues, AsyncAlchemicalQueue[T], AsyncAlchemicalTaskQueue[T, R]) built on sqlalchemy.ext.asyncio, with every I/O method async def. alchemical_queues.tasks.aio.AsyncWorker runs async def task handlers registered via async_task/AsyncTask. Table models, entries/responses and serializers are shared unchanged with the sync implementation. Tested and CI-covered against Postgres via both asyncpg and psycopg (v3 async mode).
  • Runnable mini example apps under examples/ -- plain Python, async plain Python, FastAPI, Flask, Starlette and Litestar -- each scheduling a task and polling for its result, rendered into the Examples docs so they never drift from the code they show. fastapi_app demonstrates TaskResultSerializer(PydanticSerializer(...)); flask_app demonstrates JsonSerializer.
  • Serializer: a pluggable way to control how queue entries and task responses are turned into bytes for storage and back, instead of always using pickle. Pass an instance via the new serializer= argument of AlchemicalQueues.get()/get_typed()/get_task_queue()/get_task_queue_typed(). Why: pickle is fine as a default (it's what every previous version always used, unchanged), but some applications want a human-readable/interoperable wire format, or schema validation on what goes through the queue.
  • PickleSerializer: the existing pickle-based behavior, now explicit and still the default everywhere a serializer= isn't passed.
  • JsonSerializer: serializes with the standard library json module.
  • PydanticSerializer: serializes a single pydantic model type to/from JSON, given the model class. Requires the new optional pydantic extra (pip install alchemical_queues[pydantic]) -- pydantic is never a hard dependency.
  • AlchemicalQueues.get_serialized()/get_task_queue_serialized(): like get_typed()/get_task_queue_typed(), but infer the queue's type T from the Serializer[T] you pass instead of a separate typeof argument -- e.g. queues.get_serialized("q", PydanticSerializer(MyModel)) gives you an AlchemicalQueue[MyModel], with entry.data type-checked as MyModel, without repeating the type.
  • AlchemicalTaskQueue is now Generic[T, R], with entries (put()/get(), type T) and responses (respond()/responses(), type R) independently typed and independently serialized via the new response_serializer= argument (get_task_queue()/get_task_queue_serialized()). AlchemicalResponse is now Generic[R] to match, so responses()[i].data is typed without a manual cast(). Why: a task's input and its result are usually different shapes -- sharing one serializer between them meant a non-pickle serializer= on a task queue (e.g. PydanticSerializer(Job)) made respond() unusable for anything but that same type.
  • tasks.TaskResultSerializer: a response_serializer for a queue tasks.Worker runs, which lets a task's success value go through a serializer of your choice while keeping Worker's own {"result": ...}/{"error": ..., "error_type": ...} envelope (which QueuedTask.result needs to tell success from failure) -- e.g. TaskResultSerializer(PydanticSerializer(JobResult)) makes QueuedTask.result come back as a JobResult on success, still a TaskException on failure.
  • Tested and CI-covered database/driver support for MySQL, MariaDB and SQL Server (pymssql, sync only) alongside the existing SQLite/PostgreSQL coverage -- see Database support. Oracle was attempted but pulled: it turned up widespread test failures and an indefinite hang, so it isn't supported yet (see Driver limitations).

Fixed

  • AlchemicalQueue.get() and AlchemicalTaskQueue.get() now work on MySQL and MariaDB. Both previously assumed DELETE/UPDATE ... RETURNING was available on every dialect in FOR UPDATE SKIP LOCKED's allowlist; MySQL supports no RETURNING at all, and MariaDB supports it for DELETE but not UPDATE, so every AlchemicalTaskQueue.get() call (and, on MySQL, every AlchemicalQueue.get() call too) raised instead of claiming/popping an entry. Both methods now fall back to locking the candidate row first (SELECT ... FOR UPDATE [SKIP LOCKED]) and issuing a plain UPDATE/DELETE by id in the same transaction when the dialect can't do it atomically via RETURNING.
  • The queue table's queue_name column was Text, indexed -- MySQL/MariaDB can't put an index on an unbounded TEXT/BLOB column (create_all() raised 1170 (42000): BLOB/TEXT column ... used in key specification without a key length), so AlchemicalQueues(engine=...).create_all() never worked on either. It's now String(255), which is portable across every supported backend.
  • Every timestamp column (enqueued_at, schedule_at, claimed_until, delivered_at, cleanup_at) now requests microsecond precision explicitly on MySQL/MariaDB (DATETIME(6)). Unlike PostgreSQL and SQLite, MySQL's DATETIME silently rounds to whole seconds unless asked otherwise, which broke anything comparing timestamps less than a second apart -- visibility_timeout/schedule_at under a second, extend(), and ordering entries enqueued within the same second.
  • Every write method on AlchemicalQueues/AlchemicalQueue/AlchemicalTaskQueue (and their async counterparts), including AlchemicalTaskQueue.responses()'s expired-response cleanup delete, now retries on a MySQL/MariaDB deadlock (error 1213), with a small randomized backoff between attempts, instead of letting it propagate. InnoDB's gap locking on the queue_name/priority index range scan get()/claim() does its candidate SELECT ... FOR UPDATE with can deadlock against concurrent put() inserts into the same range under heavy write concurrency; since the deadlock victim's transaction is rolled back in full before it's raised, retrying the whole call is always safe, and spacing retries out keeps heavily concurrent writers from repeatedly colliding in lockstep.
  • A queue name longer than 255 characters (the queue_name column's limit) now raises ValueError from AlchemicalQueues.get()/get_task_queue() (and the async equivalents) immediately, on every backend. Previously only PostgreSQL/MySQL/SQL Server enforced it, at write time, with a backend-specific error -- SQLite's unbounded TEXT column silently accepted it, so the same code could work on SQLite and fail elsewhere.

Version 0.2.0

Changed

  • SQLAlchemy 2.x is now required (sqlalchemy>=2.0,<3). The internal models moved from the legacy registry()/DeclarativeMeta/Column pattern to DeclarativeBase/Mapped/mapped_column.
  • AlchemicalQueue and AlchemicalTaskQueue are now separate classes. AlchemicalQueue (from queues.get()/get_typed()) stays a minimal FIFO/priority queue: put(), get() (deletes and returns the next entry outright), qsize(), empty(), clear(). Everything related to reliable task delivery -- claims, visibility_timeout, release()/discard()/extend(), respond()/responses() -- moved to the new AlchemicalTaskQueue, obtained via queues.get_task_queue()/get_task_queue_typed(). alchemical_queues.tasks.Worker now requires an AlchemicalTaskQueue. Why: most users putting plain data through a queue never touch claims or responses at all, and the claim-based get() added below bulked up every AlchemicalQueue with machinery that's only relevant to the task-queue use case. See Database support.
  • AlchemicalTaskQueue.get() no longer needs a BEGIN EXCLUSIVE transaction on SQLite, and no longer deletes the entry it returns — it claims it instead (an atomic UPDATE ... RETURNING on SQLite, with FOR UPDATE SKIP LOCKED added on PostgreSQL/MySQL/Oracle so concurrent workers don't contend on the same rows). A claimed entry stays visible to qsize()/empty() as outstanding work, and other get() calls can't claim it, until you call discard() or release() (both new, see below). This is why: previously, a worker that died between get() and respond() lost the entry silently and permanently, since the row was already gone. Now an unreleased claim simply expires after its visibility_timeout and becomes claimable again, so a crashed worker costs you a delay, not the task. alchemical_queues.tasks.Worker already does the right thing here (claim handling, retries, and keyboard interrupts all go through the same discard/release logic) — this mainly matters if you use AlchemicalTaskQueue directly.
  • AlchemicalTaskQueue.respond() no longer touches the claimed entry at all -- it only ever recorded a response, and conflating "record an outcome" with "manage the claim" turned out to hide a real bug (see Fixed, below). Call discard() once you've responded, same as tasks.Worker does.
  • The project's environment and dependencies are now managed with uv instead of Poetry.

Added

  • AlchemicalQueues accepts an optional base argument: an existing DeclarativeBase to attach its tables to, so they can share a registry and metadata with the rest of your application (e.g. Flask-SQLAlchemy's db.Model) instead of always getting an isolated one.
  • AlchemicalQueues.get_task_queue()/get_task_queue_typed(): obtain an AlchemicalTaskQueue by name, accepting a visibility_timeout to control how long a claimed entry stays claimed before becoming available again (see above).
  • AlchemicalTaskQueue.release(entry_id, claim_token): put a claimed entry back in the queue, claimable again immediately, without recording a response.
  • AlchemicalTaskQueue.discard(entry_id, claim_token): remove a claimed entry entirely without recording a response.
  • AlchemicalTaskQueue.extend(entry_id, claim_token, by=None): push a claim's expiry further out, for work that runs longer than visibility_timeout. tasks.Worker(queue, keepalive_every=...) does this automatically in a background thread for as long as a task handler is running.
  • AlchemicalEntry.claim_token: a random value fresh for each claim, required by release()/discard()/extend() alongside the entry_id. This is a deliberate fencing token: if a claim has since expired and been reclaimed by someone else (same entry_id, new claim_token), a worker that's still running late can't mistake its own stale claim for the current one -- those calls raise the new ClaimExpired instead of silently succeeding against a claim that's moved on.
  • QueuedTask.done: True once a task has a recorded outcome (success or failure). Added because task.result is None both while a task hasn't finished yet and after it finishes successfully with no return value — there was no way to tell those apart for a task whose success value is legitimately None.
  • TaskException.exception_type: the failed task's original exception class name (e.g. "ValueError"), alongside the existing stringified msg. TaskException also now has a readable __repr__/__str__.
  • alchemical_worker accepts --import module:attribute as an alternative to the positional engine URL, to run against an app-provided AlchemicalQueues instance (one using base=, custom table names, or a non-default visibility_timeout) instead of always constructing a default one — e.g. alchemical_worker --import myapp.queues:queues task-queue.

Fixed

  • On SQLite, entry_id and response_id could be silently reused once a queue emptied out and refilled, since SQLite recycles a deleted row's rowid by default. Entry and response tables now set sqlite_autoincrement, so ids stay unique for the lifetime of the database, matching the documented guarantee.
  • responses() deleted expired responses (past their cleanup_at) but never committed that delete, so it was silently rolled back when the session closed and expired responses were never actually purged.
  • A claim whose visibility_timeout lapsed while a worker was still (genuinely) working on it could be redelivered, finished by a second worker, and then have the first worker's now-stale respond() call also succeed -- recording two responses for one entry. Found with a multi-threaded chaos test against SQLite mixing successful, released, discarded, and abandoned ("crashed") claims across many workers; fixed by the claim_token fencing described above, which makes a stale worker's discard()/release()/extend() call raise ClaimExpired instead of silently succeeding. tasks.Worker now proves it still holds a task's claim (via discard()) before responding or retrying, not after, so a duplicate response can no longer happen as long as responses go through discard()-then-respond(), the same order tasks.Worker itself uses.
  • QueuedTask.result raised a raw AttributeError if the recorded response wasn't the {"result": ...}/{"error": ...} dict shape tasks.Worker itself responds with -- which respond() never required, since it's a general queue-level API independent of tasks.Worker. Calling respond() directly with anything else (a plain string, say) now has result hand that value back to you as-is instead of crashing.

Version 0.1.1

Previous stable release.

Version 0.1.0

Initial release.