cr0hn / aiotasks

A Celery like task manager that distributes Asyncio coroutines

aiotasks

aiotasks: A Celery like task manager that distributes Asyncio coroutines

Project site	https://github.com/cr0hn/aiotasks
Issues	https://github.com/cr0hn/aiotasks/issues/
Documentation	http://aiotasks.readthedocs.io
Author	Daniel Garcia (cr0hn) - @ggdaniel
Latest Version	1.0.0-alpha
Python versions	3.5 or above

Support this project

Support this project (to solve issues, new features...) by applying the Github "Sponsor" button.

What's aiotasks

aiotasks is an asynchronous & distributed task/jobs queue, implemented as coroutines and based on Python asyncio framework.

Based on the Celery Task Queue ideas, but distributing coroutines and focusing on performance, non-blocking, & event-driven concepts.

aiotasks does not pull and does not actively wait for incoming jobs. Instead it uses asyncio framework to suspend the execution until any new data is received by a new broker notification.

Documentation

You can find documentation at: https://aiotasks.readthedocs.org/

Licence

aiotasks is released under BSD license.

Contributors

Contributors are welcome. You can find a list of TODO tasks in the TODO.md at the project file.

All contributors will be added to the CONTRIBUTORS.md file.

Thanks in advance if you're planning to contribute to the project! :)

aiotasks's Issues

TypeError: DB must be of int type, not '12'

manager=build_manager("redis://192.168.1.93:6379/12")
why？

TODO

Hi @cr0hn. Can I take this tasks:

• Add doctest
• Add / fix unit testing

?

python3.9 support

not able to install on python3.9, OS Debian bullseye

Current Scenario

ashutoshpednekar@Ashutoshs-MacBook-Air ~ % docker run -it python:3.9-bullseye bash
root@39c9ba4fc18d:/# pip install aiotasks
Collecting aiotasks
  Downloading aiotasks-1.0.0.tar.gz (21 kB)
  Preparing metadata (setup.py) ... done
Collecting click
  Downloading click-8.1.3-py3-none-any.whl (96 kB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 96.6/96.6 KB 4.7 MB/s eta 0:00:00
Collecting aioredis
  Downloading aioredis-2.0.1-py3-none-any.whl (71 kB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 71.2/71.2 KB 18.9 MB/s eta 0:00:00
Collecting colorlog
  Downloading colorlog-6.6.0-py2.py3-none-any.whl (11 kB)
Collecting booby-ng
  Downloading booby-ng-0.8.4.tar.gz (14 kB)
  Preparing metadata (setup.py) ... error
  error: subprocess-exited-with-error
  
  × python setup.py egg_info did not run successfully.
  │ exit code: 1
  ╰─> [1 lines of output]
      error in booby-ng setup command: use_2to3 is invalid.
      [end of output]
  
  note: This error originates from a subprocess, and is likely not a problem with pip.
error: metadata-generation-failed

× Encountered error while generating package metadata.
╰─> See above for output.

note: This is an issue with the package mentioned above, not pip.
hint: See above for details.
WARNING: You are using pip version 22.0.4; however, version 22.1.2 is available.
You should consider upgrading via the '/usr/local/bin/python -m pip install --upgrade pip' command.

Expected:

ashutoshpednekar@Ashutoshs-MacBook-Air ~ % docker run -it python:3.7-buster bash 
root@aa37fd18dff9:/# pip install aiotasks
Collecting aiotasks
  Downloading aiotasks-1.0.0.tar.gz (21 kB)
  Preparing metadata (setup.py) ... done
Collecting click
  Downloading click-8.1.3-py3-none-any.whl (96 kB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 96.6/96.6 KB 5.5 MB/s eta 0:00:00
Collecting aioredis
  Downloading aioredis-2.0.1-py3-none-any.whl (71 kB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 71.2/71.2 KB 11.8 MB/s eta 0:00:00
Collecting colorlog
  Downloading colorlog-6.6.0-py2.py3-none-any.whl (11 kB)
Collecting booby-ng
  Downloading booby-ng-0.8.4.tar.gz (14 kB)
  Preparing metadata (setup.py) ... done
Collecting msgpack-python
  Downloading msgpack-python-0.5.6.tar.gz (138 kB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 139.0/139.0 KB 10.6 MB/s eta 0:00:00
  Preparing metadata (setup.py) ... done
Collecting async-timeout
  Downloading async_timeout-4.0.2-py3-none-any.whl (5.8 kB)
Collecting typing-extensions
  Downloading typing_extensions-4.3.0-py3-none-any.whl (25 kB)
Collecting six
  Downloading six-1.16.0-py2.py3-none-any.whl (11 kB)
Collecting importlib-metadata
  Downloading importlib_metadata-4.12.0-py3-none-any.whl (21 kB)
Collecting zipp>=0.5
  Downloading zipp-3.8.0-py3-none-any.whl (5.4 kB)
Building wheels for collected packages: aiotasks, booby-ng, msgpack-python
  Building wheel for aiotasks (setup.py) ... done
  Created wheel for aiotasks: filename=aiotasks-1.0.0-py3-none-any.whl size=39293 sha256=1124e717a4099a1e7a7c2f426f213501798b39604b655281726d9586f97829d6
  Stored in directory: /root/.cache/pip/wheels/2a/97/fd/51f1399b47f2c8976b016b6e1be8c07a98f0022220f379ef68
  Building wheel for booby-ng (setup.py) ... done
  Created wheel for booby-ng: filename=booby_ng-0.8.4-py3-none-any.whl size=17722 sha256=fe446d961511f944503b744e4417dab6644c39b5f5f935dafdf7401f02ac80fc
  Stored in directory: /root/.cache/pip/wheels/30/31/3d/e838f7415d844fe42e6a6371dc0281b2521c50b75e14da913d
  Building wheel for msgpack-python (setup.py) ... done
  Created wheel for msgpack-python: filename=msgpack_python-0.5.6-cp37-cp37m-linux_aarch64.whl size=352772 sha256=ef46c2d2b34c3d23e7704ed9f436c57c5da0331e91a1ed4df583200a958ec14d
  Stored in directory: /root/.cache/pip/wheels/f8/6c/02/92ebc97f3b99ad5bfc675be2c513f9cb3504fdbe338314f377
Successfully built aiotasks booby-ng msgpack-python
Installing collected packages: msgpack-python, zipp, typing-extensions, six, colorlog, importlib-metadata, booby-ng, async-timeout, click, aioredis, aiotasks
Successfully installed aioredis-2.0.1 aiotasks-1.0.0 async-timeout-4.0.2 booby-ng-0.8.4 click-8.1.3 colorlog-6.6.0 importlib-metadata-4.12.0 msgpack-python-0.5.6 six-1.16.0 typing-extensions-4.3.0 zipp-3.8.0
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv
WARNING: You are using pip version 22.0.4; however, version 22.1.2 is available.
You should consider upgrading via the '/usr/local/bin/python -m pip install --upgrade pip' command.

Can't stop standalone manager correctly

When I run examples/standalone_tasks_standalone.py, the program can't stop after all tasks have been consumed.

I'll open a pull request later to fix it. @cr0hn

Docs at readthedocs is broken

the readme links https://aiotasks.readthedocs.io/ as the doc page, but it raises a 404.

Is latest version published?

• Documentation at https://aiotasks.readthedocs.io/en/latest/quickstart.html#send-the-tasks seems incorrect

>>> import asyncio
>>> from aiotasks import build_manager, send_task
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
ImportError: cannot import name 'send_task' from 'aiotasks' (/usr/local/lib/python3.7/site-packages/aiotasks/__init__.py)

checking my version

bash-4.4# pip list | grep aiotasks
aiotasks         1.0.0

The distribution on my local does not have this function from source: https://github.com/cr0hn/aiotasks/blob/master/aiotasks/tasks/bases.py#L404

when i do a cat /usr/local/lib/python3.7/site-packages/aiotasks/tasks/bases.py

Is it possible to get task id?

Hi, thank you very much for the project!
I have a question. It's working with celery:
task_id = celery.send_task('task_name')
Celery send_task returns id and I can use it in other place to get result. For example:
result = celery.AsyncResult(task_id)
I tried with aiotasks, but was not successful.
task_id = await send_task("task_name")
Aiotasks send_task returns 1 always. And I don't know how to get a result by task_id.

AttributeError: module 'concurrent' has no attribute 'futures'

Hi, thanks for the project, the use case seems promising. I've been meaning to test it for a while and ended up installing python 3 to try it and run into this issue with all scripts:

$ aiotasks -vvvv worker -A examples.launch_manager_tasks_and_launch_in_console
Traceback (most recent call last):
  File "/Users/hassene/Projects/personal/spiga/env/bin/aiotasks", line 11, in <module>
    load_entry_point('aiotasks', 'console_scripts', 'aiotasks')()
  File "/Users/hassene/Projects/personal/spiga/env/lib/python3.6/site-packages/pkg_resources/__init__.py", line 570, in load_entry_point
    return get_distribution(dist).load_entry_point(group, name)
  File "/Users/hassene/Projects/personal/spiga/env/lib/python3.6/site-packages/pkg_resources/__init__.py", line 2751, in load_entry_point
    return ep.load()
  File "/Users/hassene/Projects/personal/spiga/env/lib/python3.6/site-packages/pkg_resources/__init__.py", line 2405, in load
    return self.resolve()
  File "/Users/hassene/Projects/personal/spiga/env/lib/python3.6/site-packages/pkg_resources/__init__.py", line 2411, in resolve
    module = __import__(self.module_name, fromlist=['__name__'], level=0)
  File "/Users/hassene/Projects/personal/spiga/env/lib/python3.6/site-packages/aiotasks/__init__.py", line 1, in <module>
    from .core import *
  File "/Users/hassene/Projects/personal/spiga/env/lib/python3.6/site-packages/aiotasks/core/__init__.py", line 4, in <module>
    from .exceptions import *
  File "/Users/hassene/Projects/personal/spiga/env/lib/python3.6/site-packages/aiotasks/core/exceptions.py", line 16, in <module>
    class AioTasksTimeout(concurrent.futures.TimeoutError):
AttributeError: module 'concurrent' has no attribute 'futures'

Do you have any idea what I'm doing wrong or is there something broken?

Incorrect documentation of send and wait response

Hi,

I found the documentation of this page is incorrect: https://aiotasks.readthedocs.io/en/latest/quickstart.html#sending-info-to-tasks-wait-for-response

The method should be with .delay, .wait is not exists.