Add runner for asyncio tasks (#1405)

* add asyncio runner

* add asyncio runner

* fix for old version

* fix tests
main
MyrikLD 4 years ago committed by GitHub
parent 59d1b40d14
commit 14ca7881e4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -8,6 +8,7 @@ import pickle
import warnings
import zlib
import asyncio
from collections.abc import Iterable
from distutils.version import StrictVersion
from functools import partial
@ -720,7 +721,12 @@ class Job(object):
pipeline.hmset(self.key, mapping)
def _execute(self):
return self.func(*self.args, **self.kwargs)
result = self.func(*self.args, **self.kwargs)
if asyncio.iscoroutine(result):
loop = asyncio.get_event_loop()
coro_result = loop.run_until_complete(result)
return coro_result
return result
def get_ttl(self, default_ttl=None):
"""Returns ttl for a job that determines how long a job will be

@ -32,6 +32,11 @@ def say_hello(name=None):
return 'Hi there, %s!' % (name,)
async def say_hello_async(name=None):
"""A async job with a single argument and a return value."""
return say_hello(name)
def say_hello_unicode(name=None):
"""A job with a single argument and a return value."""
return text_type(say_hello(name)) # noqa

@ -748,6 +748,21 @@ class TestJob(RQTestCase):
self.assertRaises(TypeError, queue.enqueue, fixtures.say_hello, job_id=1234)
def test_create_job_with_async(self):
"""test creating jobs with async function"""
queue = Queue(connection=self.testconn)
async_job = queue.enqueue(fixtures.say_hello_async, job_id="async_job")
sync_job = queue.enqueue(fixtures.say_hello, job_id="sync_job")
self.assertEqual(async_job.id, "async_job")
self.assertEqual(sync_job.id, "sync_job")
async_task_result = async_job.perform()
sync_task_result = sync_job.perform()
self.assertEqual(sync_task_result, async_task_result)
def test_get_call_string_unicode(self):
"""test call string with unicode keyword arguments"""
queue = Queue(connection=self.testconn)

Loading…
Cancel
Save