Close the Redis pool in the documentation examples

docs/examples/custom_serialization_msgpack.py

@@ -16,7 +16,10 @@ async def main():
         job_serializer=msgpack.packb,
         job_deserializer=lambda b: msgpack.unpackb(b, raw=False),
     )
-    await redis.enqueue_job('the_task')
+    try:
+        await redis.enqueue_job('the_task')
+    finally:
+        await redis.close()
 
 
 class WorkerSettings:

docs/examples/deferred.py

@@ -9,15 +9,17 @@ async def the_task(ctx):
 
 async def main():
     redis = await create_pool(RedisSettings())
+    try:
+        # deferred by 10 seconds
+        await redis.enqueue_job('the_task', _defer_by=10)
 
-    # deferred by 10 seconds
-    await redis.enqueue_job('the_task', _defer_by=10)
+        # deferred by 1 minute
+        await redis.enqueue_job('the_task', _defer_by=timedelta(minutes=1))
 
-    # deferred by 1 minute
-    await redis.enqueue_job('the_task', _defer_by=timedelta(minutes=1))
-
-    # deferred until jan 28th 2032, you'll be waiting a long time for this...
-    await redis.enqueue_job('the_task', _defer_until=datetime(2032, 1, 28))
+        # deferred until jan 28th 2032, you'll be waiting a long time for this...
+        await redis.enqueue_job('the_task', _defer_until=datetime(2032, 1, 28))
+    finally:
+        await redis.close()
 
 class WorkerSettings:
     functions = [the_task]

docs/examples/job_abort.py

@@ -11,9 +11,12 @@ async def do_stuff(ctx):
 
 async def main():
     redis = await create_pool(RedisSettings())
-    job = await redis.enqueue_job('do_stuff')
-    await asyncio.sleep(1)
-    await job.abort()
+    try:
+        job = await redis.enqueue_job('do_stuff')
+        await asyncio.sleep(1)
+        await job.abort()
+    finally:
+        await redis.close()
 
 
 class WorkerSettings:

docs/examples/job_ids.py

@@ -10,42 +10,44 @@ async def the_task(ctx):
 
 async def main():
     redis = await create_pool(RedisSettings())
-
-    # no id, random id will be generated
-    job1 = await redis.enqueue_job('the_task')
-    print(job1)
-    """
-    >  <arq job 99edfef86ccf4145b2f64ee160fa3297>
-    """
-
-    # random id again, again the job will be enqueued and a job will be returned
-    job2 = await redis.enqueue_job('the_task')
-    print(job2)
-    """
-    >  <arq job 7d2163c056e54b62a4d8404921094f05>
-    """
-
-    # custom job id, job will be enqueued
-    job3 = await redis.enqueue_job('the_task', _job_id='foobar')
-    print(job3)
-    """
-    >  <arq job foobar>
-    """
-
-    # same custom job id, job will not be enqueued and enqueue_job will return None
-    job4 = await redis.enqueue_job('the_task', _job_id='foobar')
-    print(job4)
-    """
-    >  None
-    """
-
-    # you can retrieve jobs by using arq.jobs.Job
-    await redis.enqueue_job('the_task', _job_id='my_job')
-    job5 = Job(job_id='my_job', redis=redis)
-    print(job5)
-    """
-    <arq job my_job>
-    """
+    try:
+        # no id, random id will be generated
+        job1 = await redis.enqueue_job('the_task')
+        print(job1)
+        """
+        >  <arq job 99edfef86ccf4145b2f64ee160fa3297>
+        """
+
+        # random id again, again the job will be enqueued and a job will be returned
+        job2 = await redis.enqueue_job('the_task')
+        print(job2)
+        """
+        >  <arq job 7d2163c056e54b62a4d8404921094f05>
+        """
+
+        # custom job id, job will be enqueued
+        job3 = await redis.enqueue_job('the_task', _job_id='foobar')
+        print(job3)
+        """
+        >  <arq job foobar>
+        """
+
+        # same custom job id, job will not be enqueued and enqueue_job will return None
+        job4 = await redis.enqueue_job('the_task', _job_id='foobar')
+        print(job4)
+        """
+        >  None
+        """
+
+        # you can retrieve jobs by using arq.jobs.Job
+        await redis.enqueue_job('the_task', _job_id='my_job')
+        job5 = Job(job_id='my_job', redis=redis)
+        print(job5)
+        """
+        <arq job my_job>
+        """
+    finally:
+        await redis.close()
 
 class WorkerSettings:
     functions = [the_task]

docs/examples/job_results.py

@@ -11,43 +11,45 @@ async def the_task(ctx):
 
 async def main():
     redis = await create_pool(RedisSettings())
-
-    job = await redis.enqueue_job('the_task')
-
-    # get the job's id
-    print(job.job_id)
-    """
-    >  68362958a244465b9be909db4b7b5ab4 (or whatever)
-    """
-
-    # get information about the job, will include results if the job has finished, but
-    # doesn't await the job's result
-    debug(await job.info())
-    """
-    >   docs/examples/job_results.py:23 main
-    JobDef(
-        function='the_task',
-        args=(),
-        kwargs={},
-        job_try=None,
-        enqueue_time=datetime.datetime(2019, 4, 23, 13, 58, 56, 781000),
-        score=1556027936781
-    ) (JobDef)
-    """
-
-    # get the Job's status
-    print(await job.status())
-    """
-    >  JobStatus.queued
-    """
-
-    # poll redis for the job result, if the job raised an exception,
-    # it will be raised here
-    # (You'll need the worker running at the same time to get a result here)
-    print(await job.result(timeout=5))
-    """
-    >  42
-    """
+    try:
+        job = await redis.enqueue_job('the_task')
+
+        # get the job's id
+        print(job.job_id)
+        """
+        >  68362958a244465b9be909db4b7b5ab4 (or whatever)
+        """
+
+        # get information about the job, will include results if the job has finished, but
+        # doesn't await the job's result
+        debug(await job.info())
+        """
+        >   docs/examples/job_results.py:25 main
+        JobDef(
+            function='the_task',
+            args=(),
+            kwargs={},
+            job_try=None,
+            enqueue_time=datetime.datetime(2019, 4, 23, 13, 58, 56, 781000),
+            score=1556027936781
+        ) (JobDef)
+        """
+
+        # get the Job's status
+        print(await job.status())
+        """
+        >  JobStatus.queued
+        """
+
+        # poll redis for the job result, if the job raised an exception,
+        # it will be raised here
+        # (You'll need the worker running at the same time to get a result here)
+        print(await job.result(timeout=5))
+        """
+        >  42
+        """
+    finally:
+        await redis.close()
 
 class WorkerSettings:
     functions = [the_task]

docs/examples/main_demo.py

@@ -21,8 +21,11 @@ async def shutdown(ctx):
 
 async def main():
     redis = await create_pool(REDIS_SETTINGS)
-    for url in ('https://facebook.com', 'https://microsoft.com', 'https://github.com'):
-        await redis.enqueue_job('download_content', url)
+    try:
+        for url in ('https://facebook.com', 'https://microsoft.com', 'https://github.com'):
+            await redis.enqueue_job('download_content', url)
+    finally:
+        await redis.close()
 
 # WorkerSettings defines the settings to use when creating the work,
 # It's used by the arq CLI.

docs/examples/retry.py

@@ -21,7 +21,10 @@ async def shutdown(ctx):
 
 async def main():
     redis = await create_pool(RedisSettings())
-    await redis.enqueue_job('download_content', 'https://httpbin.org/status/503')
+    try:
+        await redis.enqueue_job('download_content', 'https://httpbin.org/status/503')
+    finally:
+        await redis.close()
 
 class WorkerSettings:
     functions = [download_content]

docs/examples/slow_job.py

@@ -8,7 +8,10 @@ async def the_task(ctx):
 
 async def main():
     redis = await create_pool(RedisSettings())
-    await redis.enqueue_job('the_task')
+    try:
+        await redis.enqueue_job('the_task')
+    finally:
+        await redis.close()
 
 class WorkerSettings:
     functions = [the_task]

tests/test_pool_shutdown.py

@@ -0,0 +1,28 @@
+import asyncio
+import gc
+import io
+from contextlib import redirect_stderr
+
+from arq.connections import RedisSettings, create_pool
+
+
+def test_create_pool_close_before_loop_shutdown(test_redis_settings: RedisSettings) -> None:
+    async def main() -> None:
+        redis = await create_pool(test_redis_settings)
+        await redis.ping()
+        await redis.close()
+        connections = [
+            *redis.connection_pool._available_connections,
+            *redis.connection_pool._in_use_connections,
+        ]
+        assert all(getattr(connection, '_writer', None) is None for connection in connections)
+
+    loop = asyncio.new_event_loop()
+    stderr = io.StringIO()
+    try:
+        loop.run_until_complete(main())
+    finally:
+        with redirect_stderr(stderr):
+            loop.close()
+            gc.collect()
+    assert 'Event loop is closed' not in stderr.getvalue()