Use Pika like a framework?

[bencleary]

I am in the process of building out a consumer application and i am wondering if you can immediately see any issues with this type of pattern? (ignore the code quality i knocked it up in 10 minutes as a test 😅)

class AMQP:

    def __init__(self):
        credentials = pika.PlainCredentials('a_user', 'a_password')
        self.params = pika.ConnectionParameters(host="localhost", port=5671, credentials=credentials, heartbeat=10)
        self.connection = pika.BlockingConnection(parameters=self.params)
        self.channel = self.connection.channel()
    
    def subscribe(self, topic):
        @wraps(topic)
        def wrapper(handler):
            self.channel.basic_consume(topic, handler)
            return handler
        return wrapper
    
    def start(self):
        try:
            self.channel.start_consuming()
        except KeyboardInterrupt:
            self.channel.stop_consuming()
            self.connection.close()


app = AMQP()

@app.subscribe("message-events")
def printer(channel, method_frame, header_frame, body):
    print(f"Topic 1 - {body}")


@app.subscribe("events")
def printer2(channel, method_frame, header_frame, body):
    print(f"Topic 2 - {body}")

app.start()

This all works fine actually and quite nicely, i just need to implement a level of robustness. As background i am coming from using NestJS and their approach that makes a lot of sense and hides the connection details and keeps it quite nice to use. There is obviously some other considerations here with retying, etc and better parameter configuration, but just curious if there is anything screaming at you (other than the rough code 😅) like its going to implode here and here, or you will face this and that if you do X.

[lukebakken]

Looks fine to me. Just be sure to realize that when you do work when messages arrive (via handler) that the work you do can't block the main thread which runs Pika's I/O loop.