pybeef 0.0.14

Opinionated distributed RPC using AMQP workers

beef

Beef is good for your health.

An opinionated distributed RPC using AMQP messages (inspired by celery amqp backend).

Installing

pip install pybeef

Usage

Worker:

from beef import beef

@beef
async def add(x: int, y: int) -> int:
    return x + y

Server:

async def server():
    async with add.connect('amqp://localhost/'):
        await add.serve()  # here server blocks forever, executing incoming tasks

if __name__ == '__main__':
    import asyncio

    asyncio.run(server())

Client (blocking):

async def client():
    async with add.connect('amqp://localhost/'):
        task_id = await add.submit(1, 2)  # submits task
        result = await add.result(task_id)  # waits for completion

if __name__ == '__main__':
    import asyncio

    asyncio.run(client())

Client (polling):

import asyncio

async def client():
    async with add.connect('amqp://localhost/'):
        task_id = await add.submit(1, 2)  # submits task

        while True:
            status = await add.get_status(task_id)
            if status.is_final:
                break
            await asyncio.sleep(1)
        print(f'Task {task_id} completed with status {status})

if __name__ == '__main__':
    import asyncio

    asyncio.run(client())

Testing

You need a running RabbitMQ server for testing. For example:

docker run -it --network host rabbitmq

Then, make sure that test dependencies are installed:

pip install .[test]

Finally, run the tests

python -m pytest beef/test

or

make test

Building

pip wheel . --no-deps -w wheels/

or

make

Publishing

twine upload wheels/pybeef-XXX.whl -u __token__ -p <secret-pipy-token>