如何使用烧瓶 webapp 正确运行机器人答案

【问题标题】：How to properly run a bot with a flask webapp如何使用烧瓶 webapp 正确运行机器人
【发布时间】：2022-01-07 20:23:40
【问题描述】：

目标是运行一个包含多个文件的机器人脚本，需要与flask webapp共享数据库

models.py

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

class User(  db.Model ):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(100))
    pass_hash = db.Column(db.String(1000))

app.py

from flask import Flask
from models import db, User 
from test import start_bot
from threading import Thread

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///sth.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db.init_app(app)


@app.route('/')
def get_info():
    return "this is a test"

if __name__ == "__main__":
    bot_thread = Thread(target=start_bot).start()
    app.run()

bot.py这是简化为单个文件的bot的主文件

from time import time,sleep
from models import db 
from models import User

def access_db():    
    all_usrs = User.query.all()
    return (str(all_usrs))

def start_bot():
    while True:
        print (f"Current Time : {time()}")
        sleep(3)
        access_db()

现在机器人需要访问数据库并返回该数据库中所有用户的列表，并且需要从该数据库编辑添加或删除错误

RuntimeError: No application found. Either work inside a view function or push an application context. See http://flask-sqlalchemy.pocoo.org/contexts/.

机器人似乎超出了应用程序上下文我尝试使用 with app.app_context(): 运行线程，但这似乎并没有解决问题

你是怎么解决这个问题的

【问题讨论】：

标签： python flask

【解决方案1】：

好的，你可以试试这个，但我认为这不是一个好的解决方案

在app.py添加这个


def create_app():
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///sth.db"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    db.init_app(app)
    return app

将from test import start_bot 移动到 bot_thread 之前的行，否则会引发错误

ImportError: cannot import name 'create_app' from partially initialized module 'app' (most likely due to a circular import)

在bot.py 中将access_db 更改为


from app import create_app


def access_db():
    with create_app().app_context():
    
        all_usrs = User.query.all()
        return (str(all_usrs))

正如我所说，这不是最好的解决方案，但如果您遇到任何错误，请告诉我

【讨论】：