Repository: python-arq/arq (language: python)

Issue:
python 3.10 support
greetings @samuelcolvin 
I was trying to test arq on python 3.10.0 and I'm kind of going in circles :snake:  

1. arq uses aioredis 1.3.1 
2. there is a bug on python 3.10.0  (https://bugs.python.org/issue45416) which prevent that version of aioredis to function correctly `        self._cond = asyncio.Condition(lock=Lock())`

what do you think would be the best way to proceed ? 

It seems like trying to push https://github.com/samuelcolvin/arq/pull/259 is the right move however, I haven't check precisely if aioredis 2.x was hit but the python bug (for which there is a workaround) 

I could also ask if aioredis intends to fix the 1.3.1 version, which would sort the 1st blocker I'm facing on this.

let me know your thoughts, thanks !


Discussion / hints:
Just in case it's useful, I had the same problem so I patched our code with the fix (which is added in 3.10.1):
```python
def _patch_asyncio_condition():
    """
    Apply a bugfix in asyncio.Condition that is added towards Python 3.10.1.
    Once we update to Python 3.10.1, this patch can be removed.
    See:
        https://github.com/python/cpython/commit/164dddf5f8c9c6b93f32c9f79b4301fc804576e9
    The bug affects an internal call to asyncio.Condition in `arq` which is used by this module.
    This function will raise a RuntimeError once we upgrade to 3.10.1, and should be removed.
    Raises:
        RuntimeError: If the Python version is not 3.10.0 and the function should be removed.
    """

    def _patched_condition_init(self, lock=None, *, loop=asyncio.mixins._marker):  # type: ignore
        super(type(self), self).__init__(loop=loop)
        if lock is None:
            lock = asyncio.Lock()

        self._lock = lock
        # Export the lock's locked(), acquire() and release() methods.
        self.locked = lock.locked
        self.acquire = lock.acquire
        self.release = lock.release

        self._waiters = collections.deque()

    if platform.python_version() == "3.10.0":
        asyncio.Condition.__init__ = _patched_condition_init  # type: ignore
    else:
        raise RuntimeError("Looks like Python was updated, remove the patch to asyncio in this module.")
```


Fix patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index fb0f71b5..d382c822 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -49,7 +49,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu]
-        python-version: ['3.7', '3.8', '3.9']
+        python-version: ['3.7', '3.8', '3.9', '3.10']
 
     env:
       PYTHON: ${{ matrix.python-version }}
diff --git a/setup.py b/setup.py
index 5c5e6f2c..904c06a5 100644
--- a/setup.py
+++ b/setup.py
@@ -34,6 +34,7 @@
         'Programming Language :: Python :: 3.7',
         'Programming Language :: Python :: 3.8',
         'Programming Language :: Python :: 3.9',
+        'Programming Language :: Python :: 3.10',
         'Topic :: Internet',
         'Topic :: Software Development :: Libraries :: Python Modules',
         'Topic :: System :: Clustering',


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 0417ad6c..a7282068 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -11,7 +11,6 @@
 
 @pytest.fixture(name='loop')
 def _fix_loop(event_loop):
-    asyncio.set_event_loop(event_loop)
     return event_loop
 
 
diff --git a/tests/test_cli.py b/tests/test_cli.py
index d49033f3..55ca6b8f 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -20,7 +20,8 @@ def test_help():
     assert result.output.startswith('Usage: arq [OPTIONS] WORKER_SETTINGS\n')
 
 
-def test_run(cancel_remaining_task):
+def test_run(cancel_remaining_task, mocker, loop):
+    mocker.patch('asyncio.get_event_loop', lambda: loop)
     runner = CliRunner()
     result = runner.invoke(cli, ['tests.test_cli.WorkerSettings'])
     assert result.exit_code == 0
diff --git a/tests/test_worker.py b/tests/test_worker.py
index e3e7ab0a..7f376b13 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -33,7 +33,7 @@ async def fails(ctx):
     raise TypeError('my type error')
 
 
-def test_no_jobs(arq_redis: ArqRedis, loop):
+def test_no_jobs(arq_redis: ArqRedis, loop, mocker):
     class Settings:
         functions = [func(foobar, name='foobar')]
         burst = True
@@ -41,7 +41,7 @@ class Settings:
         queue_read_limit = 10
 
     loop.run_until_complete(arq_redis.enqueue_job('foobar'))
-    asyncio.set_event_loop(loop)
+    mocker.patch('asyncio.get_event_loop', lambda: loop)
     worker = run_worker(Settings)
     assert worker.jobs_complete == 1
     assert str(worker) == '<Worker j_complete=1 j_failed=0 j_retried=0 j_ongoing=0>'