aiormq

Pure python AMQP asynchronous client library

These details have not been verified by PyPI

Project links

Homepage

Project description

https://img.shields.io/pypi/wheel/aiormq.svg

https://img.shields.io/pypi/pyversions/aiormq.svg

https://img.shields.io/pypi/l/aiormq.svg

aiormq is a pure python AMQP client library.

Status

3.x.x branch - Production/Stable
4.x.x branch - Unstable (Experimental)

Features

Connecting by URL

amqp example: amqp://user:password@server.host/vhost

secure amqp example: amqps://user:password@server.host/vhost?cafile=ca.pem&keyfile=key.pem&certfile=cert.pem&no_verify_ssl=0

Buffered queue for received frames
Only PLAIN auth mechanism support
Publisher confirms support
Transactions support
Channel based asynchronous locks

Note

AMQP 0.9.1 requires serialize sending for some frame types on the channel. e.g. Content body must be following after content header. But frames might be sent asynchronously on another channels.
Tracking unroutable messages (Use connection.channel(on_return_raises=False) for disabling)
Full SSL/TLS support, using your choice of:
- amqps:// url query parameters:
  
  cafile= - string contains path to ca certificate file
  
  capath= - string contains path to ca certificates
  
  cadata= - base64 encoded ca certificate data
  
  keyfile= - string contains path to key file
  
  certfile= - string contains path to certificate file
  
  no_verify_ssl - boolean disables certificates validation
- context= SSLContext keyword argument to connect().
Python type hints
Uses pamqp as an AMQP 0.9.1 frame encoder/decoder

Tutorial

Introduction

Simple consumer

import asyncio
import aiormq

async def on_message(message):
    """
    on_message doesn't necessarily have to be defined as async.
    Here it is to show that it's possible.
    """
    print(" [x] Received message %r" % message)
    print("Message body is: %r" % message.body)
    print("Before sleep!")
    await asyncio.sleep(5)   # Represents async I/O operations
    print("After sleep!")


async def main():
    # Perform connection
    connection = await aiormq.connect("amqp://guest:guest@localhost/")

    # Creating a channel
    channel = await connection.channel()

    # Declaring queue
    declare_ok = await channel.queue_declare('helo')
    consume_ok = await channel.basic_consume(
        declare_ok.queue, on_message, no_ack=True
    )


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_forever()

Simple publisher

import asyncio
import aiormq


async def main():
    # Perform connection
    connection = await aiormq.connect("amqp://guest:guest@localhost//")

    # Creating a channel
    channel = await connection.channel()

    # Sending the message
    await channel.basic_publish(b'Hello World!', routing_key='hello')
    print(" [x] Sent 'Hello World!'")


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Work Queues

Create new task

import sys
import asyncio
import aiormq


async def main():
    # Perform connection
    connection = await aiormq.connect("amqp://guest:guest@localhost/")

    # Creating a channel
    channel = await connection.channel()

    body = b' '.join(sys.argv[1:]) or b"Hello World!"

    # Sending the message
    await channel.basic_publish(
        body,
        routing_key='task_queue',
        properties=aiormq.spec.Basic.Properties(
            delivery_mode=1,
        )
    )

    print(" [x] Sent %r" % body)

    await connection.close()


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Simple worker

import asyncio
import aiormq
import aiormq.types


async def on_message(message: aiormq.types.DeliveredMessage):
    print(" [x] Received message %r" % (message,))
    print("     Message body is: %r" % (message.body,))


async def main():
    # Perform connection
    connection = await aiormq.connect("amqp://guest:guest@localhost/")


    # Creating a channel
    channel = await connection.channel()
    await channel.basic_qos(prefetch_count=1)

    # Declaring queue
    declare_ok = await channel.queue_declare('task_queue', durable=True)

    # Start listening the queue with name 'task_queue'
    await channel.basic_consume(declare_ok.queue, on_message, no_ack=True)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

# we enter a never-ending loop that waits for data and runs
# callbacks whenever necessary.
print(" [*] Waiting for messages. To exit press CTRL+C")
loop.run_forever()

Publish Subscribe

Publisher

import sys
import asyncio
import aiormq


async def main():
    # Perform connection
    connection = await aiormq.connect("amqp://guest:guest@localhost/")

    # Creating a channel
    channel = await connection.channel()

    await channel.exchange_declare(
        exchange='logs', exchange_type='fanout'
    )

    body = b' '.join(sys.argv[1:]) or b"Hello World!"

    # Sending the message
    await channel.basic_publish(
        body, routing_key='info', exchange='logs'
    )

    print(" [x] Sent %r" % (body,))

    await connection.close()


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Subscriber

import asyncio
import aiormq
import aiormq.types


async def on_message(message: aiormq.types.DeliveredMessage):
    print("[x] %r" % (message.body,))

    await message.channel.basic_ack(
        message.delivery.delivery_tag
    )


async def main():
    # Perform connection
    connection = await aiormq.connect("amqp://guest:guest@localhost/")

    # Creating a channel
    channel = await connection.channel()
    await channel.basic_qos(prefetch_count=1)

    await channel.exchange_declare(
        exchange='logs', exchange_type='fanout'
    )

    # Declaring queue
    declare_ok = await channel.queue_declare(exclusive=True)

    # Binding the queue to the exchange
    await channel.queue_bind(declare_ok.queue, 'logs')

    # Start listening the queue with name 'task_queue'
    await channel.basic_consume(declare_ok.queue, on_message)


loop = asyncio.get_event_loop()
loop.create_task(main())

# we enter a never-ending loop that waits for data
# and runs callbacks whenever necessary.
print(' [*] Waiting for logs. To exit press CTRL+C')
loop.run_forever()

Routing

Direct consumer

import sys
import asyncio
import aiormq
import aiormq.types


async def on_message(message: aiormq.types.DeliveredMessage):
    print(" [x] %r:%r" % (message.delivery.routing_key, message.body))
    await message.channel.basic_ack(
        message.delivery.delivery_tag
    )


async def main():
    # Perform connection
    connection = aiormq.Connection("amqp://guest:guest@localhost/")
    await connection.connect()

    # Creating a channel
    channel = await connection.channel()
    await channel.basic_qos(prefetch_count=1)

    severities = sys.argv[1:]

    if not severities:
        sys.stderr.write(
            "Usage: %s [info] [warning] [error]\n" % sys.argv[0]
        )
        sys.exit(1)

    # Declare an exchange
    await channel.exchange_declare(
        exchange='logs', exchange_type='direct'
    )

    # Declaring random queue
    declare_ok = await channel.queue_declare(durable=True, auto_delete=True)

    for severity in severities:
        await channel.queue_bind(
            declare_ok.queue, 'logs', routing_key=severity
        )

    # Start listening the random queue
    await channel.basic_consume(declare_ok.queue, on_message)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

# we enter a never-ending loop that waits for data
# and runs callbacks whenever necessary.
print(" [*] Waiting for messages. To exit press CTRL+C")
loop.run_forever()

Emitter

import sys
import asyncio
import aiormq


async def main():
    # Perform connection
    connection = await aiormq.connect("amqp://guest:guest@localhost/")

    # Creating a channel
    channel = await connection.channel()

    await channel.exchange_declare(
        exchange='logs', exchange_type='direct'
    )

    body = (
        b' '.join(arg.encode() for arg in sys.argv[2:])
        or
        b"Hello World!"
    )

    # Sending the message
    routing_key = sys.argv[1] if len(sys.argv) > 2 else 'info'

    await channel.basic_publish(
        body, exchange='logs', routing_key=routing_key,
        properties=aiormq.spec.Basic.Properties(
            delivery_mode=1
        )
    )

    print(" [x] Sent %r" % body)

    await connection.close()


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Topics

Publisher

import sys
import asyncio
import aiormq


async def main():
    # Perform connection
    connection = await aiormq.connect("amqp://guest:guest@localhost/")

    # Creating a channel
    channel = await connection.channel()

    await channel.exchange_declare('topic_logs', exchange_type='topic')

    routing_key = (
        sys.argv[1] if len(sys.argv) > 2 else 'anonymous.info'
    )

    body = (
        b' '.join(arg.encode() for arg in sys.argv[2:])
        or
        b"Hello World!"
    )

    # Sending the message
    await channel.basic_publish(
        body, exchange='topic_logs', routing_key=routing_key,
        properties=aiormq.spec.Basic.Properties(
            delivery_mode=1
        )
    )

    print(" [x] Sent %r" % (body,))

    await connection.close()


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Consumer

import asyncio
import sys
import aiormq
import aiormq.types


async def on_message(message: aiormq.types.DeliveredMessage):
    print(" [x] %r:%r" % (message.delivery.routing_key, message.body))
    await message.channel.basic_ack(
        message.delivery.delivery_tag
    )


async def main():
    # Perform connection
    connection = await aiormq.connect(
        "amqp://guest:guest@localhost/", loop=loop
    )

    # Creating a channel
    channel = await connection.channel()
    await channel.basic_qos(prefetch_count=1)

    # Declare an exchange
    await channel.exchange_declare('topic_logs', exchange_type='topic')

    # Declaring queue
    declare_ok = await channel.queue_declare('task_queue', durable=True)

    binding_keys = sys.argv[1:]

    if not binding_keys:
        sys.stderr.write(
            "Usage: %s [binding_key]...\n" % sys.argv[0]
        )
        sys.exit(1)

    for binding_key in binding_keys:
        await channel.queue_bind(
            declare_ok.queue, 'topic_logs', routing_key=binding_key
        )

    # Start listening the queue with name 'task_queue'
    await channel.basic_consume(declare_ok.queue, on_message)


loop = asyncio.get_event_loop()
loop.create_task(main())

# we enter a never-ending loop that waits for
# data and runs callbacks whenever necessary.
print(" [*] Waiting for messages. To exit press CTRL+C")
loop.run_forever()

Remote procedure call (RPC)

RPC server

import asyncio
import aiormq
import aiormq.types


def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n-1) + fib(n-2)


async def on_message(message:aiormq.types.DeliveredMessage):
    n = int(message.body.decode())

    print(" [.] fib(%d)" % n)
    response = str(fib(n)).encode()

    await message.channel.basic_publish(
        response, routing_key=message.header.properties.reply_to,
        properties=aiormq.spec.Basic.Properties(
            correlation_id=message..header.properties.correlation_id
        ),

    )

    await message.channel.basic_ack(message.delivery.delivery_tag)
    print('Request complete')


async def main():
    # Perform connection
    connection = await aiormq.connect("amqp://guest:guest@localhost/")

    # Creating a channel
    channel = await connection.channel()

    # Declaring queue
    declare_ok = await channel.queue_declare('rpc_queue')

    # Start listening the queue with name 'hello'
    await channel.basic_consume(declare_ok.queue, on_message)


loop = asyncio.get_event_loop()
loop.create_task(main())

# we enter a never-ending loop that waits for data
# and runs callbacks whenever necessary.
print(" [x] Awaiting RPC requests")
loop.run_forever()

RPC client

import asyncio
import uuid
import aiormq
import aiormq.types


class FibonacciRpcClient:
    def __init__(self):
        self.connection = None      # type: aiormq.Connection
        self.channel = None         # type: aiormq.Channel
        self.callback_queue = ''
        self.futures = {}
        self.loop = loop

    async def connect(self):
        self.connection = await aiormq.connect("amqp://guest:guest@localhost/")

        self.channel = await self.connection.channel()
        declare_ok = await self.channel.queue_declare(
            exclusive=True, auto_delete=True
        )

        await self.channel.basic_consume(declare_ok.queue, self.on_response)

        self.callback_queue = declare_ok.queue

        return self

    async def on_response(self, message: aiormq.types.DeliveredMessage):
        future = self.futures.pop(message..header.properties.correlation_id)
        future.set_result(message.body)

    async def call(self, n):
        correlation_id = str(uuid.uuid4())
        future = loop.create_future()

        self.futures[correlation_id] = future

        await self.channel.basic_publish(
            str(n).encode(), routing_key='rpc_queue',
            properties=aiormq.spec.Basic.Properties(
                content_type='text/plain',
                correlation_id=correlation_id,
                reply_to=self.callback_queue,
            )
        )

        return int(await future)


async def main():
    fibonacci_rpc = await FibonacciRpcClient().connect()
    print(" [x] Requesting fib(30)")
    response = await fibonacci_rpc.call(30)
    print(" [.] Got %r" % response)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Project details

These details have not been verified by PyPI

Project links

Homepage

Release history Release notifications | RSS feed

6.8.1

Sep 4, 2024

6.8.0

Jan 15, 2024

6.7.7

Jul 26, 2023

6.7.6

May 12, 2023

6.7.5

May 11, 2023

6.7.4

Mar 10, 2023

6.7.3

Mar 4, 2023

6.7.2

Feb 18, 2023

6.7.1

Feb 10, 2023

6.7.0

Feb 9, 2023

6.6.4

Dec 30, 2022

6.6.3

Dec 15, 2022

6.6.2

Dec 15, 2022

6.6.1

Dec 15, 2022

6.6.0

Dec 15, 2022

6.5.0

Dec 13, 2022

6.4.2

Sep 13, 2022

6.4.1

Aug 11, 2022

6.4.1b0 pre-release

Aug 11, 2022

6.4.0

Jul 19, 2022

6.3.4

Jul 12, 2022

6.3.3

May 19, 2022

6.3.2

May 16, 2022

6.3.1

May 16, 2022

6.3.0

May 16, 2022

6.2.3

Feb 25, 2022

6.2.2

Feb 9, 2022

6.2.1

Feb 9, 2022

6.2.0

Feb 9, 2022

6.1.1

Jan 11, 2022

This version

6.1.0

Nov 8, 2021

6.0.0

Nov 2, 2021

5.2.2

Jun 11, 2021

5.2.1

Apr 19, 2021

5.2.0

Apr 17, 2021

5.1.0

Apr 16, 2021

5.0.0

Apr 16, 2021

4.3.1

Apr 12, 2021

4.3.0

Apr 12, 2021

4.2.1

Mar 10, 2021

4.2.0

Feb 26, 2021

4.1.1

Dec 29, 2020

4.0.1

Nov 16, 2020

4.0.0

Nov 16, 2020

3.3.1

Oct 15, 2020

3.3.0

Oct 13, 2020

3.2.3

Jul 28, 2020

3.2.2

May 6, 2020

3.2.1

Feb 17, 2020

3.2.0

Nov 20, 2019

3.1.2

Nov 19, 2019

3.1.1

Nov 19, 2019

3.1.0

Nov 13, 2019

3.0.1

Oct 30, 2019

2.9.1

Oct 16, 2019

2.9.0

Oct 16, 2019

2.8.1

Oct 16, 2019

2.8.0

Sep 23, 2019

2.7.5

Sep 5, 2019

2.7.4

Aug 29, 2019

2.7.3

Aug 22, 2019

2.7.2

Aug 2, 2019

2.7.1

Aug 1, 2019

2.7.0

Aug 1, 2019

2.6.1b2 pre-release

Jul 4, 2019

2.6.1b0 pre-release

Jul 1, 2019

2.6.0

Jun 28, 2019

2.5.6

Jun 21, 2019

2.5.5

Jun 13, 2019

2.5.4

Jun 13, 2019

2.5.3

May 14, 2019

2.5.2

Apr 30, 2019

2.5.1

Apr 23, 2019

2.5.0

Apr 19, 2019

2.4.2

Apr 19, 2019

2.4.1

Apr 15, 2019

2.4.0

Apr 13, 2019

2.3.3

Mar 26, 2019

2.3.1

Mar 24, 2019

2.1.1

Mar 24, 2019

2.1.0

Mar 10, 2019

2.0.1

Feb 26, 2019

2.0.0

Feb 26, 2019

1.2.2

Feb 26, 2019

1.2.1

Feb 21, 2019

1.2.0

Feb 20, 2019

1.1.2

Feb 18, 2019

1.1.1

Feb 11, 2019

1.1.0

Feb 11, 2019

1.0.2

Feb 11, 2019

1.0.1

Feb 8, 2019

1.0.0

Feb 7, 2019

0.9.0

Feb 7, 2019

0.8.0

Feb 4, 2019

0.7.3

Feb 3, 2019

0.7.2

Jan 22, 2019

0.7.1

Jan 21, 2019

0.7.0

Jan 20, 2019

0.6.1

Jan 20, 2019

0.6.0

Jan 20, 2019

0.5.0

Jan 20, 2019

0.4.0

Jan 20, 2019

0.3.3

Jan 20, 2019

0.3.2

Jan 10, 2019

0.3.1

Jan 10, 2019

0.3.0

Jan 10, 2019

0.2.3

Jan 8, 2019

0.2.2

Jan 8, 2019

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aiormq-6.1.0.tar.gz (26.4 kB view details)

Uploaded Nov 8, 2021 Source

Built Distribution

aiormq-6.1.0-py3-none-any.whl (33.0 kB view details)

Uploaded Nov 8, 2021 Python 3

File details

Details for the file aiormq-6.1.0.tar.gz.

File metadata

Download URL: aiormq-6.1.0.tar.gz
Upload date: Nov 8, 2021
Size: 26.4 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/3.4.1 importlib_metadata/3.10.0 pkginfo/1.7.0 requests/2.25.1 requests-toolbelt/0.9.1 tqdm/4.59.0 CPython/3.9.2

File hashes

Hashes for aiormq-6.1.0.tar.gz
Algorithm	Hash digest
SHA256	`d7c6877ec1a614319f60abd16221ba31caeac909fcbfb119d0b13176461243e5`
MD5	`e7b59c724123dc6f6a8718b55443c07e`
BLAKE2b-256	`f323e442f942aa2407c7db09cb065888e95e3c8880dc7268ad512f5975919497`

See more details on using hashes here.

File details

Details for the file aiormq-6.1.0-py3-none-any.whl.

File metadata

Download URL: aiormq-6.1.0-py3-none-any.whl
Upload date: Nov 8, 2021
Size: 33.0 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/3.4.1 importlib_metadata/3.10.0 pkginfo/1.7.0 requests/2.25.1 requests-toolbelt/0.9.1 tqdm/4.59.0 CPython/3.9.2

File hashes

Hashes for aiormq-6.1.0-py3-none-any.whl
Algorithm	Hash digest
SHA256	`9724bcba114030378503073feffa49f9b2b8a8fcb6986cb81189ec42b80a78cb`
MD5	`42ff3f46af017b2c0f64fc376393be37`
BLAKE2b-256	`78435bbaf347d35add9fd500521fdc122cda60d4322d15786ec41bcf21047734`

See more details on using hashes here.

aiormq 6.1.0

Navigation

Verified details

Maintainers

Unverified details

Project links

Meta

Classifiers

Project description

Project details

Verified details

Maintainers

Unverified details

Project links

Meta

Classifiers

Release history Release notifications | RSS feed

Download files

Source Distribution

Built Distribution

File details

File metadata

File hashes

File details

File metadata

File hashes