如何在 Flask 应用程序 URL 端点之外使用 SQLAlchemy 查询 Flask SQLite 数据库？

Question

我正在创建一个 Flask REST API。它作为 API 运行良好。我越来越需要连接到应用程序的 sqlite 数据库来执行计划任务（这篇文章可以解决），还需要通过 MQTT 消息连接到其他后台物联网触发的日志。

到目前为止，当我尝试导入为在 Flask 应用程序中使用而创建的 ORM 模型时，我收到此错误：

ImportError: attempted relative import with no known parent package

这是有道理的，因为模型正在扩展 Flask 的数据库 [会话？，什么？]。这是一个计划任务示例，我首先尝试重新创建 Flask 应用程序环境以使用模型并运行任务（也许我不需要此处的完整 Flask 应用程序来使用模型？）：

删除_旧_tokens.py

# 🔥 I was going to import model here, yet that generates error.
# from ..models import TokenBlocklist

from flask import Flask
from datetime import datetime, timedelta
import os
from flask_sqlalchemy import SQLAlchemy

# TODO: Not sure I like this added globally, yet maybe useful to bring in from
# another file for these types of tasks.
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' \
    + os.path.abspath('../../instance/db.sqlite')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.app_context().push()
db = SQLAlchemy(app)
db.create_all()


# Tokens are set to expire at max 30 days. After about that long with padding,
# remove the tokens from the database to keep that table clean.
def remove_old_tokens():
    # 🔥 This line generates the same error as just importing at top of the file
    from ..models import TokenBlocklist

    forty_days = timedelta(days=40)
    forty_days_ago = datetime.now() - forty_days
    query = TokenBlocklist.__table__.delete() \
        .where(TokenBlocklist.created < forty_days_ago)
    db.session.execute(query)
    db.session.commit()
    print('old tokens deleted')


remove_old_tokens()

模型.py

import uuid
from .app import db


def uuid_str():
    return str(uuid.uuid4())



class TokenBlocklist(db.Model):
    id = db.Column(
        db.String(36),
        primary_key=True,
        nullable=False,
        index=True,
        default=uuid_str
    )
    jti = db.Column(
        db.String(36),
        nullable=False,
        index=True
    )
    type = db.Column(
        db.String(10),
        nullable=False
    )
    created_at = db.Column(
        db.DateTime,
        nullable=False,
        server_default=func.now(),
        index=True
    )

应用程序.py

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db.sqlite'
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db = SQLAlchemy(app)
with app.app_context():
    db.create_all()

结构

app/
    app.py
    models.py
    scheduled_tasks/
        remove_old_tokens.py
instance/
    db.sqlite

我能找到的一种替代方法是编写简单的文本查询，忽略模型（找到很好的例子here）。然而，如果当应用程序增长时我不再使用 SQLite，这种情况最终可能会被打破。

如何解决该错误？请记住，在使用数据库时，我所做的不仅仅是可以使用 URL 的计划任务（例如，UI 将使用的实时 IoT 日志）。

完整代码（不包括正在制定的示例）可在此处获得。

更新：

感谢 @michael-butscher 的评论，我能够使用使用绝对导入的更改脚本解决导入错误。但是，现在我遇到了循环导入错误。我想知道重新设计模型文件是否有意义，或者只是以另一种方式查询：

ImportError: cannot import name 'TokenBlocklist' from partially initialized module 'app.models' (most likely due to a circular import) (/full-path-here/app/models.py)

import sys
import os
sys.path.append(os.path.abspath('../../'))

from flask import Flask
from datetime import datetime, timedelta
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.abspath('../../instance/db.sqlite')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# 🔥 This may be the issue, because models.py is trying to load db from un-initialized app.py...
db = SQLAlchemy(app)
app.app_context().push()

# 🔥 app.models is going to try `from .app import db` hmmm...
from app.models import TokenBlocklist
db.create_all()


def remove_old_tokens():
    forty_days = timedelta(days=40)
    forty_days_ago = datetime.now() - forty_days
    query = TokenBlocklist.__table__.delete().where(TokenBlocklist.created < forty_days_ago)
    db.session.execute(query)
    db.session.commit()
    print('old tokens deleted')

remove_old_tokens()

Answer 1

感谢@michael-butscher 的评论，我能够使用使用绝对导入的更改脚本解决导入错误。之后，我打破了应用程序对数据库的使用，以便主 Flask 应用程序和此脚本都可以使用模型。

新：数据库.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

模型.py

...
from .database import db
...

删除_旧_tokens.py

from flask import Flask
from datetime import datetime, timedelta
import sys
import os

sys.path.append(os.path.abspath('../../'))
from app.database import db
from app.models import TokenBlocklist


def remove_old_tokens():
    forty_days = timedelta(days=40)
    forty_days_ago = datetime.now() - forty_days
    query = TokenBlocklist.__table__.delete().where(
        TokenBlocklist.created_at < forty_days_ago
    )
    db.session.execute(query)
    db.session.commit()
    print('old tokens deleted')


app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' \
    + os.path.abspath('../../instance/db.sqlite')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db.init_app(app)
with app.app_context():
    db.create_all()
    remove_old_tokens()

⚠️ 如果有一种方法可以在不涉及 Flask 的情况下运行此查询，我会接受这是一个更好的答案。