Pika Type Stubs
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
-
ModuleNotFoundError: No module named 'pika'
cause You have installed `types-pika` but forgotten to install the actual `pika` library, which contains the runtime code.fixInstall the `pika` library: `pip install pika` (or `pip install -e '.[dev]'` if in a project). -
error: Missing type information for "pika" [attr-defined] in your_script.py
cause MyPy or your chosen type checker cannot find the `types-pika` stubs, even if installed. This might be due to incorrect installation or MyPy configuration.fixFirst, ensure `types-pika` is installed (`pip install types-pika`). If it is, verify your MyPy configuration allows it to find third-party stubs (e.g., `mypy --install-types --non-interactive` to automatically install missing stubs, or check `follow_imports` settings in `mypy.ini`).
Warnings
- gotcha types-pika is a stub-only package. It provides type hints for the 'pika' library but does not include the Pika runtime code itself. You must install both `types-pika` AND `pika` for your application to run and for type checking to be effective.
- gotcha Static type checkers like MyPy need to be configured correctly to discover and use type stub packages. If MyPy reports 'Missing type information for "pika"' even after installing `types-pika`, your MyPy configuration might be too lenient or not configured to follow imports properly.
Install
-
pip install types-pika pika
Imports
- BlockingConnection
from pika import BlockingConnection
- URLParameters
from pika import URLParameters
- channel.Channel
import pika.channel
Quickstart
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)