Python Flask Application Context，如何将应用程序暴露给其他模块

Question

我有一个使用 Flask-SqlAlchemy 访问 Postgres 数据库的 python 脚本。但是，每当我尝试查询数据库时，都会收到“脱离上下文”错误。我认为这样做的方法是将其包装在 app.app_context 中：

import psycopg2
import json
from simple_chalk import redBright
from ...models.bin import Bin
from ...models.station import Station
from ... import db
from datetime import datetime as dt
from ... import current_app as app

def findPositionBin(stationId, find_position):
    try:
        with app.app_context():
            result = Bin.query.filter_by(station_id=stationId).filter_by(position=find_position).first()
            print("result")
            return result
    except Exception as e:
        print(redBright(e))

但是，为此我需要导入 app.问题是我的根 init.py 将应用程序包含在要由 wsgi.py 调用以运行程序的函数中。

初始化.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_socketio import SocketIO
from flask_cors import CORS
import eventlet
import threading


db = SQLAlchemy()
migrate = Migrate()
socketio = SocketIO()


def create_app():
    app = Flask(__name__, instance_relative_config=False)
    CORS(app)
    app.config.from_object('config.Config')
    eventlet.monkey_patch()
    socketio.init_app(app, cors_allowed_origins='*', async_mode='eventlet')
    migrate.init_app(app, db)

    with app.app_context():
        from . import routes
        from . import wsroutes
        from .models import user, bin, ip_port, station
        from .blueprints import user
        from .blueprints.CubeStation import station_routes
        from.database.CubeStation import station_database
        from .server import startServer
        from .blueprints.CubeStation.station_functions import broadcastLoop
        # from .database.CubeStation import station_database
        db.init_app(app)
        app.register_blueprint(user.user_blueprint)
        app.register_blueprint(station_routes.station_blueprint)
        # app.register_blueprint(station_database.database_blueprint)
        x = threading.Thread(target=startServer)
        x.start()
        t = threading.Thread(target=broadcastLoop)
        t.start()
        db.create_all()

        return app

有谁知道我如何公开应用程序以便其他模块可以导入它？或者，如果有更好的方法来解决这个问题。提前致谢

Answer 1

基于here 和here，也许这种方式会起作用：

from init import app

def app_context():
    with app.app_context():
        yield


def findPositionBin(stationId, find_position, app_context):
    try:
        with app.app_context():
            result = Bin.query.filter_by(station_id=stationId).filter_by(position=find_position).first()
            print("result")
            return result
    except Exception as e:
        print(redBright(e))