wiji 0.3.0

1. As a library

import asyncio
import wiji

class AdderTask(wiji.task.Task):
    the_broker = wiji.broker.InMemoryBroker()
    queue_name = "AdderTaskQueue1"

    async def run(self, a, b):
        result = a + b
        print("\nresult: {0}\n".format(result))
        return result

# queue some tasks
myAdderTask = AdderTask( )
myAdderTask.synchronous_delay(a=4, b=37)
myAdderTask.synchronous_delay(a=67, b=847)

# run the workers
worker = wiji.Worker(the_task=myAdderTask)
asyncio.run(worker.consume_tasks())

2. As a cli app

wiji also ships with a commandline app called wiji-cli.

import wiji

class AdderTask(wiji.task.Task):
    the_broker = wiji.broker.InMemoryBroker()
    queue_name = "AdderTaskQueue1"

    async def run(self, a, b):
        res = a + b
        print()
        print("res:: ", res)
        print()
        return res

MyAppInstance = wiji.app.App(task_classes=[AdderTask])

wiji-cli --app examples.my_app.MyAppInstance

Writing tests

Lets say you have wiji tasks in your project and you want to write integration or unit tests for them and their use.

# my_tasks.py

import wiji
import MyRedisBroker # a custom broker using redis

DATABASE = {}

class AdderTask(wiji.task.Task):
    the_broker = MyRedisBroker()
    queue_name = "AdderTask"

    async def run(self, a, b):
        """
        adds two numbers and stores the resut in a database
        """
        result = a + b
        DATABASE["result"] = result
        return result

class ExampleView:
    def post(self, request):
        a = request["a"]
        b = request["b"]
        AdderTask().synchronous_delay(a=a, b=b)

# test_tasks.py

from my_tasks import ExampleView
from unittest import TestCase

class TestExampleView(TestCase):
    def test_view(self):
        view = ExampleView()
        view.post(request={"a": 45, "b": 46})
        # do your asserts here

# test_tasks.py

import asyncio
from my_tasks import ExampleView, AdderTask, DATABASE
from unittest import TestCase, mock

class TestExampleView(TestCase):

    @staticmethod
    def _run(coro):
        """
        helper function that runs any coroutine in an event loop.
        see:: https://blog.miguelgrinberg.com/post/unit-testing-asyncio-code
        """
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(coro)

    def test_view(self):
        with mock.patch.object(
            # ie, substitute the redis broker with an in-memory one during test runs
            AdderTask, "the_broker", wiji.broker.InMemoryBroker()
        ) as mock_broker:
            view = ExampleView()
            view.post(request={"a": 45, "b": 46})
            # do your asserts here

    def test_whole_flow(self):
        with mock.patch.object(
            AdderTask, "the_broker", wiji.broker.InMemoryBroker()
        ) as mock_broker:
            # 1. assert that the database is initially empty
            self.assertDictEqual(DATABASE, {})

            # 2. when `post` is called it will queue a task
            view = ExampleView()
            view.post(request={"a": 45, "b": 46})

            # 3. we need to run workers
            worker = wiji.Worker(the_task=AdderTask())
            self._run(worker.consume_tasks(TESTING=True))

            # 4. assert that database has been updated succesfully.
            self.assertDictEqual(DATABASE, {"result": 91})