
在开发 flask 应用时,我们经常需要执行一些独立于 web 请求的数据库操作,例如定时清理任务、数据导入脚本或响应外部事件(如 iot 消息)的后台处理。然而,直接在这些外部脚本中导入 flask 应用中定义的 sqlalchemy 模型和数据库实例,常常会导致 importerror 或循环导入等问题。这是因为 flask-sqlalchemy 依赖于 flask 应用上下文,并且在模块导入时存在特定的初始化顺序。
最初尝试在外部脚本中直接导入模型时,会遇到 ImportError: attempted relative import with no known parent package。这是因为外部脚本作为顶级模块运行时,无法正确解析相对导入路径。即使通过调整 sys.path 解决了相对导入问题,将 Flask 应用的数据库初始化逻辑复制到外部脚本中,并尝试导入模型,又会遇到 ImportError: cannot import name 'TokenBlocklist' from partially initialized module 'app.models' (most likely due to a circular import)。
出现循环导入的原因通常是:
为了解决这些问题,我们需要一种更健壮的方式来管理 Flask-SQLAlchemy 实例,使其既能在 Flask 应用内部正常工作,也能在外部脚本中独立配置和使用。
解决上述问题的关键在于解耦 Flask-SQLAlchemy 实例的创建与初始化。我们将 SQLAlchemy() 实例的创建与它绑定到特定 Flask 应用的过程分开。
首先,创建一个名为 database.py 的新文件,专门用于实例化 SQLAlchemy 对象。这个文件只负责创建 db 对象,而不将其绑定到任何 Flask 应用。
# app/database.py from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy()
现在,models.py 不再需要从 app.py 导入 db,而是从新创建的 database.py 导入 db 实例。这打破了 models.py 对 app.py 的直接依赖。
# app/models.py
import uuid
from sqlalchemy.sql import func # 确保 func 被导入,如果模型中使用了它
from .database import db # 从独立的 database.py 导入 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(), # 使用 func.now()
index=True
)在主 Flask 应用文件 app.py 中,我们需要导入 db 实例,然后使用 db.init_app(app) 方法将其绑定到 Flask 应用实例上。
# app/app.py
from flask import Flask
from app.database import db # 从独立的 database.py 导入 db
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db.sqlite'
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db.init_app(app) # 将 db 实例与 Flask 应用绑定
with app.app_context():
db.create_all() # 在应用上下文内创建所有数据库表现在,外部脚本 remove_old_tokens.py 可以通过以下方式正确地访问数据库和模型:
# scheduled_tasks/remove_old_tokens.py
from flask import Flask
from datetime import datetime, timedelta
import sys
import os
# 确保 Python 解释器能够找到 app 模块
# 将项目根目录添加到 sys.path
# 假设项目结构为:
# project_root/
# app/
# app.py
# database.py
# models.py
# instance/
# db.sqlite
# scheduled_tasks/
# remove_old_tokens.py
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))
from app.database import db # 导入解耦的 db 实例
from app.models import TokenBlocklist # 导入模型
def remove_old_tokens():
"""
清理数据库中过期的 TokenBlocklist 记录。
"""
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('旧令牌已清理')
# 在外部脚本中创建一个最小的 Flask 应用实例
# 这是为了提供一个应用上下文,供 Flask-SQLAlchemy 使用
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' \
+ os.path.abspath(os.path.join(os.path.dirname(__file__), '../../instance/db.sqlite'))
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 将 db 实例与这个临时的 Flask 应用绑定
db.init_app(app)
# 在应用上下文内执行数据库操作
with app.app_context():
# 确保数据库表已创建(首次运行或测试时有用)
# 在生产环境中,通常由主应用负责 db.create_all()
db.create_all()
remove_old_tokens() # 调用清理函数通过将 SQLAlchemy 实例的创建与应用绑定过程解耦,我们成功地解决了在 Flask 应用外部访问数据库时遇到的导入错误和循环依赖问题。这种方法提高了代码的模块化程度和灵活性,使得 Flask-SQLAlchemy 数据库能够更方便地被后台任务、独立脚本或其他非 Web 请求场景所利用。遵循上述步骤和最佳实践,可以构建出健壮且易于维护的 Flask 应用及相关工具。
以上就是Flask-SQLAlchemy 数据库在应用外部的独立访问指南的详细内容,更多请关注php中文网其它相关文章!
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移,垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是,许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号