Install & Compatibility
Where this runs
tested against v1.2.0b1 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.910 runs
installs and imports cleanly · install 0.0s · import 0.275s · 19.4MB
glibcpy 3.10–3.910 runs
installs and imports cleanly · install 1.7s · import 0.246s · 20MB
18MB installed
● package 18MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
BlockingConnection
✓ from pika import BlockingConnection
types-pika provides type hints for Pika's symbols; you import from pika itself, not types-pika.
URLParameters
✓ from pika import URLParameters
types-pika provides type hints for Pika's symbols; you import from pika itself, not types-pika.
channel.Channel
✓ import pika.channel
types-pika provides type hints for Pika's symbols; you import from pika itself, not types-pika.
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)
Upgrade
Version history
1.2.0b1latest on PyPI · released Jan 18, 2022
Audit
Dependencies
pikarequiredProvides type stubs for the pika library; pika must be installed for runtime functionality.