Python Flask应用程序上下文，如何向其他模块公开应用程序

Question

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

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))

然而，要做到这一点，我需要导入应用程序。问题是我的rootinit.py在wsgi.py调用以运行程序的函数中包含了应用程序

init.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))