Pika Type Stubs

1.2.0b1 · active · verified Fri Apr 17

types-pika provides PEP-484 compliant type stubs for the Pika AMQP client library, enabling static type checking tools like MyPy to validate Pika usage. It is currently at version 1.2.0b1 and is actively maintained, with releases typically coinciding with or following updates to the Pika library itself.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates a basic Pika message publishing function with type hints. `types-pika` ensures that static type checkers like MyPy can validate the types used for `connection`, `channel`, and other Pika objects and methods, catching potential type errors before runtime.

import pika
from pika import BlockingConnection, URLParameters
from pika.channel import Channel

def publish_message(queue_name: str, message: str, amqp_url: str) -> None:
    """Publishes a message to a RabbitMQ queue using Pika with type hints."""
    try:
        params: URLParameters = URLParameters(amqp_url)
        connection: BlockingConnection = BlockingConnection(params)
        channel: Channel = connection.channel()

        channel.queue_declare(queue=queue_name, durable=True)
        channel.basic_publish(exchange='', routing_key=queue_name, body=message)
        print(f" [x] Sent '{message}' to queue '{queue_name}'")
    except pika.exceptions.AMQPConnectionError as e:
        print(f"Error connecting to RabbitMQ: {e}")
    finally:
        if 'connection' in locals() and connection.is_open:
            connection.close()

# Example usage (replace with your actual AMQP URL and queue)
# To run this, ensure you have a RabbitMQ instance running
# and `pika` and `types-pika` installed.
# amqp_url = os.environ.get('RABBITMQ_URL', 'amqp://guest:guest@localhost:5672/%2F')
# if __name__ == '__main__':
#     publish_message('my_test_queue', 'Hello, Pika!', amqp_url)

view raw JSON →