arrlio

Arrlio [WIP]

Documentation (WIP)

Asyncio distributed task/workflow system with supports generators and graphs

Installation

pip install arrlio

Or to use latest develop version

pip install git+https://github.com/levsh/arrlio

Usage

Create tasks file

# tasks.py

import io

import arrlio
import invoke

@arrlio.task
async def hello_world():
    return "Hello World!"

# task custom name
@arrlio.task(name="foo")
async def foo():
    arrlio.logger.info("Hello from task 'foo'!")

# exception example
@arrlio.task
async def exception():
    raise ZeroDivisionError

# Arrlio supports generators and async generators
@arrlio.task
def xrange(count):
    for x in range(count):
        yield x

@arrlio.task
async def add_one(value: str):
    return int(value) + 1


@arrlio.task
async def bash(cmd, stdin: str = None):
    in_stream = io.StringIO(stdin)
    out_stream = io.StringIO()
    result = invoke.run(
        cmd,
        in_stream=in_stream,
        out_stream=out_stream
    )
    return result.stdout

Create main file and run it

import asyncio
import logging

import arrlio
import tasks

logger = logging.getLogger("arrlio")
logger.setLevel("INFO")

BACKEND = "arrlio.backends.local"
# BACKEND = "arrlio.backends.rabbitmq"

async def main():
    app = arrlio.App(arrlio.Config(backend={"module": BACKEND}))

    async with app:
        await app.consume_tasks()

        # call by task object
        ar = await app.send_task(tasks.hello_world)
        logger.info(await ar.get())

        # call by task name
        ar = await app.send_task("foo")
        logger.info(await ar.get())

        # task args example
        ar = await app.send_task(tasks.add_one, args=(1,))
        logger.info(await ar.get())

        # exception example
        try:
            ar = await app.send_task(tasks.exception)
            logger.info(await ar.get())
        except Exception as e:
            print(f"\nThis is example exception for {app.backend}:\n")
            logger.exception(e)
            print()

        # generator example
        results = []
        ar = await app.send_task(tasks.xrange, args=(3,))
        async for result in ar:
            results.append(result)
        logger.info(results)  # -> [0, 1, 2]


if __name__ == "__main__":
    asyncio.run(main())

Arrlio supports graph execution

import asyncio
import logging

import arrlio
import tasks

logger = logging.getLogger("arrlio")
logger.setLevel("INFO")

BACKEND = "arrlio.backends.local"
# BACKEND = "arrlio.backends.rabbitmq"


async def main():
    graph = arrlio.Graph("My Graph")
    graph.add_node("A", tasks.add_one, root=True)
    graph.add_node("B", tasks.add_one)
    graph.add_node("C", tasks.add_one)
    graph.add_edge("A", "B")
    graph.add_edge("B", "C")

	# arrlio.plugins.events and arrlio.plugins.graphs
	# plugins are required
    app = arrlio.App(
        arrlio.Config(
            backend={"module": BACKEND},
            plugins=[
                {"module": "arrlio.plugins.events"},
                {"module": "arrlio.plugins.graphs"},
            ],
        )
    )

    async with app:
        await app.consume_tasks()

		# execute graph with argument 0
        ars = await app.send_graph(graph, args=(0,))
        logger.info("A: %i", await ars["A"].get())  # -> A: 1
        logger.info("B: %i", await ars["B"].get())  # -> B: 2
        logger.info("C: %i", await ars["C"].get())  # -> C: 3


if __name__ == "__main__":
    asyncio.run(main())

Another graph example

import asyncio
import logging

import arrlio
import tasks

logger = logging.getLogger("arrlio")
logger.setLevel("INFO")

BACKEND = "arrlio.backends.local"
# BACKEND = "arrlio.backends.rabbitmq"


async def main():
    graph = arrlio.Graph("My Graph")
    graph.add_node("A", tasks.bash, root=True)
    graph.add_node("B", tasks.bash, args=("wc -w",))
    graph.add_edge("A", "B")

    app = arrlio.App(
        arrlio.Config(
            backend={"module": BACKEND},
            plugins=[
                {"module": "arrlio.plugins.events"},
                {"module": "arrlio.plugins.graphs"},
            ],
        )
    )

    async with app:
        await app.consume_tasks()

        ars = await app.send_graph(
            graph,
            args=('echo "Number of words in this sentence:"',)
        )
        logger.info(await asyncio.wait_for(ars["B"].get(), timeout=2))  # -> 6


if __name__ == "__main__":
    asyncio.run(main())

And more examples

poetry install
poetry run python examples/main.py