Running worker in async function in Python 3.14

[sbusch-mobivention]

Given the following set of files:

jobs.py

import asyncio


async def test_job():
    await asyncio.sleep(0.1)
    print("test")

main.py

from redis import Redis
from rq import Queue

from jobs import test_job

redis = Redis.from_url("redis://localhost:6379")
queue = Queue(connection=redis)

if __name__ == "__main__":
    queue.enqueue(test_job)

worker.py

import asyncio

from rq import Worker

from main import queue

async def run_worker():
    worker = Worker([queue])
    worker.work(burst=True)

asyncio.run(run_worker())

Running first python main.py and then python worker.py works totally fine on python 3.13. Using python 3.14 however you get the following when running the worker:

Exception in callback <_asyncio.TaskStepMethWrapper object at 0x10623dba0>()
handle: <Handle <_asyncio.TaskStepMethWrapper object at 0x10623dba0>()>
Traceback (most recent call last):
  File "/Users/sbusch/.asdf/installs/python/3.14.0/lib/python3.14/asyncio/events.py", line 94, in _run
    self._context.run(self._callback, *self._args)
    ~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
RuntimeError: Cannot enter into task <Task pending name='Task-2' coro=<test_job() running at /XXX/rq-test/jobs.py:4> cb=[_run_until_complete_cb() at /XXX/python/3.14.0/lib/python3.14/asyncio/base_events.py:181]> while another task <Task pending name='Task-1' coro=<run_worker() running at /XXX/rq-test/worker.py:9> cb=[_run_until_complete_cb() at /XXX/python/3.14.0/lib/python3.14/asyncio/base_events.py:181]> is being executed.

Now I do realize that this is really not something you ever really want to do, however we do have some integration tests in a fully async app that enqueue tasks and run the worker in the same test step and right now it is really not possible to run those anymore without doing major hacky workarounds.