aiotarantool-queue-python's Introduction

Tarantool Queue bindings for work with python asyncio

Bindings require tarantool version 1.6 and aiotarantool connector:

$ pip install aiotarantool_queue aiotarantool

Try it example:

import asyncio
import aiotarantool_queue
import random

async def put_job(queue):
    for tube_name in ("tube1", "tube2", "tube3"):
        tube = queue.tube(tube_name)
        task = await tube.put({"task_data": random.random()})

async def take_job(tube):
    while True:
        task = await tube.take(5)
        if not task:
            break

        print(task.data)
        await task.ack()

loop = asyncio.get_event_loop()

queue = aiotarantool_queue.Queue("127.0.0.1", 3301)
put_tasks = [loop.create_task(put_job(queue))
             for _ in range(20)]

take_tasks = [loop.create_task(take_job(queue.tube(tube_name)))
              for tube_name in ("tube1", "tube2", "tube3")]

loop.run_until_complete(asyncio.wait(put_tasks + take_tasks))
loop.run_until_complete(queue.close())
loop.close()

This code makes it easy to develop your application to work with queue.

aiotarantool-queue-python's People

Stargazers

Watchers

aiotarantool-queue-python's Issues

Add support for tarantool 1.7

line from tarantool.const import ENCODING_DEFAULT fails with ImportError due to latest version of tarantool python lib

Add ability to change encoding in aiotarantool.connect

Hi, i can't change enconding on aiotarantool_queue.Queue init. If i try to put protobuf in queue, I have error from msgpack. For now i have to patch this init. So if i change encoding to None i have no errors.

def custom__init(self, host="localhost", port=33013, user=None, password=None, loop=None, lua_queue_name="box.queue"):
    if not host or not port:
        raise self.BadConfigException(
            "host and port params must be not empty"
        )

    if not isinstance(port, int):
        raise self.BadConfigException("port must be int")

    self.host = host
    self.port = port
    self.user = user
    self.password = password
    self.loop = loop or asyncio.get_event_loop()
    self.lua_queue_name = lua_queue_name
    self.tubes = dict()
    self.tnt = aiotarantool.connect(self.host, self.port,
                                    user=self.user,
                                    password=self.password,
                                    loop=self.loop, encoding=None)


aiotarantool_queue.Queue.__init__ = custom__init