pika_stubs_typed_rabbitmq_producer_quickstart.py

1import pika
2from pika.adapters.blocking_connection import BlockingChannel
3
4# pika-stubs allows type checkers to verify the connection and channel objects
5connection: pika.BlockingConnection = pika.BlockingConnection(
6    pika.ConnectionParameters(host='localhost')
7)
8channel: BlockingChannel = connection.channel()
9
10# The stubs provide autocompletion and type checking for methods like queue_declare
11channel.queue_declare(queue='hello')
12
13channel.basic_publish(
14    exchange='',
15    routing_key='hello',
16    body='Hello World!'
17)
18
19print(" [x] Sent 'Hello World!'")
20
21connection.close()