diff --git a/docs/examples/custom_serialization_msgpack.py b/docs/examples/custom_serialization_msgpack.py index b8a9f058..d6b6792f 100644 --- a/docs/examples/custom_serialization_msgpack.py +++ b/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: diff --git a/docs/examples/deferred.py b/docs/examples/deferred.py index 8b8a80cf..1ee44a09 100644 --- a/docs/examples/deferred.py +++ b/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] diff --git a/docs/examples/job_abort.py b/docs/examples/job_abort.py index 441a83eb..1e6bd54c 100644 --- a/docs/examples/job_abort.py +++ b/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: diff --git a/docs/examples/job_ids.py b/docs/examples/job_ids.py index 1148591c..2a216c93 100644 --- a/docs/examples/job_ids.py +++ b/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) - """ - > - """ - - # random id again, again the job will be enqueued and a job will be returned - job2 = await redis.enqueue_job('the_task') - print(job2) - """ - > - """ - - # custom job id, job will be enqueued - job3 = await redis.enqueue_job('the_task', _job_id='foobar') - print(job3) - """ - > - """ - - # 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) - """ - - """ + try: + # no id, random id will be generated + job1 = await redis.enqueue_job('the_task') + print(job1) + """ + > + """ + + # random id again, again the job will be enqueued and a job will be returned + job2 = await redis.enqueue_job('the_task') + print(job2) + """ + > + """ + + # custom job id, job will be enqueued + job3 = await redis.enqueue_job('the_task', _job_id='foobar') + print(job3) + """ + > + """ + + # 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) + """ + + """ + finally: + await redis.close() class WorkerSettings: functions = [the_task] diff --git a/docs/examples/job_results.py b/docs/examples/job_results.py index a8a9dc2e..89c67aa8 100644 --- a/docs/examples/job_results.py +++ b/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] diff --git a/docs/examples/main_demo.py b/docs/examples/main_demo.py index 93686187..c33a3c03 100644 --- a/docs/examples/main_demo.py +++ b/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. diff --git a/docs/examples/retry.py b/docs/examples/retry.py index 8118ecc6..1665db50 100644 --- a/docs/examples/retry.py +++ b/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] diff --git a/docs/examples/slow_job.py b/docs/examples/slow_job.py index 2bc336fe..40409297 100644 --- a/docs/examples/slow_job.py +++ b/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] diff --git a/tests/test_pool_shutdown.py b/tests/test_pool_shutdown.py new file mode 100644 index 00000000..2068f6ff --- /dev/null +++ b/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()