pika-stubs
Pika-stubs provides PEP-484 compliant type stubs and a Mypy plugin for the Pika RabbitMQ client library. Its purpose is to offer more precise static type checking and improved type inference for code written with Pika. The current version is 0.1.3, released in June 2020. The project appears to be in an unmaintained or abandoned state, classified as '1 - Planning' on PyPI with no subsequent releases.
Common errors
-
ERROR: Could not find a version that satisfies the requirement pika-stubs (from versions: none) ERROR: No matching distribution found for pika-stubs
cause The Python version being used is older than the minimum requirement (Python 3.8) for `pika-stubs`.fixUpgrade your Python environment to version 3.8 or newer. For example, `pyenv install 3.9.18` and `pyenv local 3.9.18` or `python3.9 -m pip install pika-stubs`. -
mypy: `pika` is not installed, or mypy could not find stubs for it. Please install `pika` or `pika-stubs`.
cause Either the `pika` library itself or the `pika-stubs` package is not installed, or Mypy is not configured to find them.fixEnsure both `pika` and `pika-stubs` are installed in the same environment as Mypy: `pip install pika pika-stubs`. If using a custom `mypy.ini` or setup, verify `mypy_path` or `plugins` configuration.
Warnings
- breaking Pika-stubs requires Python 3.8 or newer. Attempting to install on older Python versions (e.g., 3.7) will result in an installation failure.
- gotcha This library (pika-stubs) appears to be unmaintained. Its last release was in June 2020, and its PyPI status is '1 - Planning'. More actively maintained alternatives like `types-pika-ts` (part of typeshed) or `types-pika` may offer more up-to-date and complete typings for recent Pika versions.
Install
-
pip install pika-stubs
Imports
- BlockingConnection
import pika # Or from pika import BlockingConnection
- BasicProperties
from pika.spec import BasicProperties
Quickstart
# consumer.py
import pika
import os
def callback(ch: pika.channel.Channel, method: pika.spec.Basic.Deliver, properties: pika.spec.BasicProperties, body: bytes) -> None:
print(f" [x] Received {body.decode()}")
def main() -> None:
rabbitmq_host = os.environ.get('RABBITMQ_HOST', 'localhost')
connection = pika.BlockingConnection(
pika.ConnectionParameters(host=rabbitmq_host)
)
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
try:
channel.start_consuming()
except KeyboardInterrupt:
channel.stop_consuming()
finally:
connection.close()
if __name__ == '__main__':
main()
# To run mypy with pika-stubs:
# 1. pip install pika pika-stubs mypy
# 2. RABBITMQ_HOST=localhost python consumer.py (in one terminal)
# 3. mypy consumer.py (in another terminal)