RuntimeError: Working outside of application context.

[bsherifi]

I am having an issue with the Flask application context. I am not able to successfully run either one of the methods (however, the logs below are for the update() functionality)

queue_1    | Started Consuming...
queue_1    | Received in app
queue_1    | {'id': 40, 'title': 'new title 1', 'image': 'new image 1', 'likes': 0}
queue_1    | Traceback (most recent call last):
queue_1    |   File "/app/consumer.py", line 42, in <module>
queue_1    |     channel.start_consuming()
queue_1    |   File "/usr/local/lib/python3.10/site-packages/pika/adapters/blocking_connection.py", line 1883, in start_consuming
queue_1    |     self._process_data_events(time_limit=None)
queue_1    |   File "/usr/local/lib/python3.10/site-packages/pika/adapters/blocking_connection.py", line 2044, in _process_data_events
queue_1    |     self.connection.process_data_events(time_limit=time_limit)
queue_1    |   File "/usr/local/lib/python3.10/site-packages/pika/adapters/blocking_connection.py", line 851, in process_data_events
queue_1    |     self._dispatch_channel_events()
queue_1    |   File "/usr/local/lib/python3.10/site-packages/pika/adapters/blocking_connection.py", line 567, in _dispatch_channel_events
queue_1    |     impl_channel._get_cookie()._dispatch_events()
queue_1    |   File "/usr/local/lib/python3.10/site-packages/pika/adapters/blocking_connection.py", line 1510, in _dispatch_events
queue_1    |     consumer_info.on_message_callback(self, evt.method,
queue_1    |   File "/app/consumer.py", line 26, in callback
queue_1    |     product = Product.query.get(data['id'])
queue_1    |   File "/usr/local/lib/python3.10/site-packages/flask_sqlalchemy/model.py", line 31, in __get__
queue_1    |     cls, session=cls.__fsa__.session()  # type: ignore[arg-type]
queue_1    |   File "/usr/local/lib/python3.10/site-packages/sqlalchemy/orm/scoping.py", line 47, in __call__
queue_1    |     sess = self.registry()
queue_1    |   File "/usr/local/lib/python3.10/site-packages/sqlalchemy/util/_collections.py", line 1006, in __call__
queue_1    |     key = self.scopefunc()
queue_1    |   File "/usr/local/lib/python3.10/site-packages/flask_sqlalchemy/session.py", line 81, in _app_ctx_id
queue_1    |     return id(app_ctx._get_current_object())  # type: ignore[attr-defined]
queue_1    |   File "/usr/local/lib/python3.10/site-packages/werkzeug/local.py", line 513, in _get_current_object
queue_1    |     raise RuntimeError(unbound_message) from None
queue_1    | RuntimeError: Working outside of application context.
queue_1    | 
queue_1    | This typically means that you attempted to use functionality that needed
queue_1    | the current application. To solve this, set up an application context
queue_1    | with app.app_context(). See the documentation for more information.
main_queue_1 exited with code 1

This is my current consumer.py file state:

import json
import pika

from app import Product, db

params = pika.URLParameters('amqps://liqpanjb:nJNG1CtrG93Ian5IPfYxf5FvSk5o-OAM@moose.rmq.cloudamqp.com/liqpanjb')

connection = pika.BlockingConnection(params)

channel = connection.channel()
channel.queue_declare(queue='app')


def callback(ch, method, properties, body):
    print('Received in app')
    data = json.loads(body)
    print(data)

    if properties.content_type == 'product_created':
        product = Product(id=data['id'], title=data['title'], image=data['image'])
        db.session.add(product)
        db.session.commit()

    elif properties.content_type == 'product_updated':
        product = Product.query.get(data['id'])
        product.title = data['title']
        product.image = data['image']
        db.session.commit()

    elif properties.content_type == 'product_deleted':
        product = Product.query.get(data['id'])
        db.session.delete(product)
        db.session.commit()


channel.basic_consume(queue='app', on_message_callback=callback, auto_ack=True)

print('Started Consuming...')

channel.start_consuming()
channel.close()

This is my app.py (main) state:

from dataclasses import dataclass

from flask import Flask, jsonify
from flask_cors import CORS
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import UniqueConstraint


app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:root@db/shop_db_main'
CORS(app)

db = SQLAlchemy(app)
migrate = Migrate(app, db)


@dataclass
class Product(db.Model):
    id: int
    title: str
    image: str

    id = db.Column(db.Integer, primary_key=True, autoincrement=False)
    title = db.Column(db.String(200))
    image = db.Column(db.String(200))


@dataclass
class ProductUser(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer)
    product_id = db.Column(db.Integer)

    UniqueConstraint('user_id', 'product_id', name='user_product_unique')


@app.route('/api/products')
def index():
    return jsonify(Product.query.all())


if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0')

These are the packages I have installed (versions included):

alembic==1.8.1
certifi==2022.9.24
charset-normalizer==2.1.1
click==8.1.3
Flask==2.2.2
Flask-Cors==3.0.10
Flask-Migrate==4.0.0
Flask-Script==2.0.6
Flask-SQLAlchemy==3.0.2
greenlet==2.0.1
idna==3.4
itsdangerous==2.1.2
Jinja2==3.1.2
Mako==1.2.4
MarkupSafe==2.1.1
mysqlclient==2.1.1
pika==1.3.1
requests==2.28.1
six==1.16.0
SQLAlchemy==1.4.44
urllib3==1.26.12
Werkzeug==2.2.2

Am I overlooking something because this seems very strange behavior and I am having trouble finding a solution to it. Any help is much appreciated!

[djination]

im looking for the answer too

[ko5yak]

HERE IS A FIX
change consumer.py to this

import json
import pika

from main import app, Product, db

params = pika.URLParameters("amqps://udcupqsb:QzDC9jOai0EEcjL-d-zl-3PC2euHpMb9@shrimp.rmq.cloudamqp.com/udcupqsb")

connection = pika.BlockingConnection(params)

channel = connection.channel()

channel.queue_declare(queue="main")

def callback(channel, method, properties, body):
    print("Received in main")
    try:
        data = json.loads(body)
        print(data)
    except json.decoder.JSONDecodeError as e:
        print(f"Error decoding JSON: {e}")
        # Handle the error (e.g., log, send a response, etc.)
        return
    

    with app.app_context():
        if properties.content_type == "product_created":
            product = Product(id=data["id"], title=data["title"], image=data["image"])
            db.session.add(product)
            db.session.commit()

        elif properties.content_type == "product_updated":
            product = Product.query.get(data["id"])
            if product:
                product.title = data["title"]
                product.image = data["image"]
                db.session.commit()

        elif properties.content_type == "product_deleted":
            product = Product.query.get(data["id"])
            if product:
                db.session.delete(product)
                db.session.commit()

channel.basic_consume(queue="main", on_message_callback=callback, auto_ack=True)

print("Started consuming...")

channel.start_consuming()

channel.close()