joegasewicz/pytask-io

Python Async Task Queue

Asynchronous Tasks Library using asyncio

An Asyncio based task queue that is designed to be super easy to use!

Read the docs: Documentation

Install

pip install pytask-io

docker run redis  # Rabbit MQ coming soon...

Usage

    from pytask_io import PyTaskIO
    
    # Starts the task runner
    pytask = PytaskIO(
        store_port=6379,
        store_host="localhost",
        db=0,
        workers=1,
    )
    
    # Start the PytaskIO task queue on a separate thread.
    pytask.run()
    
    # Handle a long running process, in this case a send email function
    metadata = pytask.add_task(send_email, title, body)
    
    # Try once to get the results of your email sometime in the future
    result = pytask.get_task(metadata)
    
    # Later we can use the `metadata` result to pass to `add_task`
    result = pytask.poll_for_task(metadata, tries=100, interval=60)
    
    # Stop PytaskIO completely (This will not affect any units of work that haven't yet executed)
    pytask.stop()

Compatible task types

PyTaskIO will always return back the task meta data. Do not embed Python objects of type frame, generator, traceback & context objects. In this case you will get back the exception thrown when PyTaskIO attempts to execute the serialization.

Authors

• joegasewicz - Initial work - @joegasewicz

Contributing

Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.

Please make sure to update tests as appropriate.

License

MIT

pytask-io's Issues

result_exec_date not being set

result_exec_date not being set when the unit of work task is executed.

Docs failing

Docs failing

TestPyTaskIO::test_poll_for_task unit test fails

TestPyTaskIO::test_poll_for_task unit test fails

Docker Container image for PytaskIO

Make a PyTaskIO Docker image available to download from the Docker registry.

Set up pipenv

Set up pipenv

TestPyTaskIO::test_worker unit test fails

test_worker unit test fails

RabbitMQ as a message broker compatibility

RabbitMQ as a message broker compatibility

Redis store integration

Redis store integration

Setup Sphinx

Setup Sphinx

get_task public method

get_task public method

Store data with a unique id as a key value pair

Currently data is stored in a list with random index's

Fix: Store data with a unique id as a key value pair

TestPyTaskIO::test_add_unit_of_work unit test fails

test_add_unit_of_work unit test fails

Setup tox to run tests down to python 3.5 if possible

Setup tox to run tests down to python 3.5 if possible

stop() method blocks main thread forever

Steps to reproduce:

>>> from pytask_io import PyTaskIO
>>> pytask = PyTaskIO()
>>> pytask.run()
>>> pytask.stop()

Close asyncio loop in teardown

Close asyncio loop in teardown test method.

Not handled exception when unit of work throws an exception

Not handled exception when unit of work throws an exception:

pytask-io/pytask_io/worker.py

Line 27 in 0a917e3

executed_uow = await current_loop.run_in_executor(

Check to see if main loop attribute has been created before referencing

Check to see if main loop attribute has been created before referencing,
otherwise an AttributeError is thrown

Fix `test_add_task` tests

Looks like this tests needs fixing..
https://github.com/joegasewicz/pytask_io/blob/1921fc4b0aee177a6974492d6340128afbaab42e/tests/test_pytask_io.py#L55

Travis CI setup

Travis CI setup

Add exception when trying to access queues before run() was called.

Add usage examples for frameworks like Django, Flask, etc.

As title says Maybe add a sandbox Docker image?

Remove unused packages

There a few unused packages , such as toolz etc that need removing from the project

metadata should have an uow_error key to log errors

metadata should have an uow_error key to log errors

Codacy setup

Codacy setup

test_init should pull 1 item from the queue

Test test_init is not pulling an item from the queue when only 1 task in Redis queue

Public interface for WSGI applications

Add to the project an interface to WSGI protocol.

Docs & Readme stop method not explained

The PyTaskIO.stop() method needs to be explained in the docs

pole_for_store_results rename

rename pole_for_store_results to poll_for_store_results

Initiate PytaskIO with a run method

Should start the event loop with:

pytask.run()

Redis integration message broker integration

Redis message broker integration

Mixed licenses

mixed licenses there, LICENSE.txt is GPLv3 but README says it's MIT

(thanks to u/cetme308 on reddit)

remove `.idea from repo`

remove .idea from repo

PytaskIO incorrect import path

Currently to install PyTaskIO class we need to import like

from pytask_io.pytask_io import PyTaskIO

But we should have:

from pytask_io import PyTaskIO

Add init method

To make PyTaskIO class DI compatible add the following:

    def init(self, **kwargs):
        """
        :param kwargs:
        :return:
        """
        self.store_host = kwargs.get("store_host") or self.store_host
        self.store_port = kwargs.get("store_port") or self.store_port
        self.store_db = kwargs.get("store_db") or self.store_db
        self.workers = kwargs.get("workers") or self.workers

README.md wrong redis port

README.md wrong redis port

poll_for_task public method

poll_for_task public method

Migrate main asyncio event loop to a tcp server

Migrate main asyncio event loop to a tcp server.

By default we should automatically call stop after all items in the queue are processed

By default we should automatically call stop after all items in the queue are processed.

Unless a user chooses to keep running forever (theres should be an option for this) and let the user call stop.

Update pipenv to python 3.10

Add tools to keep the code style consistent

As title says, tools like black and isort help to keep the code style consistent. Maybe add a pre-commit file to make it more like run-and-forget.

Low level logging variable not in Readme or docs

The low level logging variable PYTASKIO_DEBUG is not in the Readme or docs.

example should read:

# For development of low level libraries, you can get debug level logging with:
export PYTASKIO_DEBUG=1

Later we can add in a kwarg to the main class for high level users...

TestPyTaskIO::test_init unit test fails

TestPyTaskIO::test_init unit test fails

Allow users of this library to choose between a Redis store & in memory

Allow users of this library to choose between a Redis store & in memory

Lets build this together!!

I need your help to build this amazing new library .. no more complex , bloated task runners... lets get this right!

Please submit issues .. pullrequests .. let's start building together , today!

setup pylint

setup pylint

Exception thrown: cannot schedule new futures after interpreter shutdown

Exception thrown: cannot schedule new futures after interpreter shutdown

All seriealizer must issue a warning for context as units of work

The serialisers can only handle certian object, specifically not context objects.
Therefore a warning should be issued and a null uof work store in the redis store with an error message

Postgres store option

Postgres store option

Remove unused store function

pytask-io/pytask_io/store.py

Line 13 in 794c747

def _connect_to_store(host: str = "localhost", port: int = 6379, db: int = 0) -> redis.Redis: