Flask应用SQLAlchemy与Twilio调度器的循环导入及定时任务排查

我是编程新手，正在搭建集成SQLAlchemy数据库的Flask应用，需要用Twilio实现每日17点发送定时短信。但在main.py或__init__.py中运行时，遭遇循环导入问题，且定时任务无法正常工作，怀疑这与Flask-SQLAlchemy和蓝图的循环导入有关。

项目目录结构 #

website
    __init__.py
    main.py
    auth.py
    keys.py
    models.py
    views.py
    static
       # Java代码（此处省略具体内容）
    templates
       # 多个HTML模板（此处省略具体内容）

相关代码 #

main.py #

if __name__ == '__main__':
    app = create_app()
    app.run(debug=True)

models.py #

from . import db
from flask_login import UserMixin
from sqlalchemy.sql import func

class User(db.Model, UserMixin):
    # 定义类属性/列
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(150), unique=True)
    password = db.Column(db.String(150))
    first_name = db.Column(db.String(150))
    phone_number = db.Column(db.String(20))
    gratitude_text_opt_in_status = db.Column(db.Boolean)

init.py #

from flask_sqlalchemy import SQLAlchemy
from os import path
from flask_login import LoginManager
from flask import Flask

# 创建SQLAlchemy对象
db = SQLAlchemy()
# 设置数据库名称
DB_NAME = "database.db"

# 定义创建数据库的函数
def create_database(app):
    # 如果数据库文件不存在
    if not path.exists('website/' + DB_NAME):
        # 使用Flask应用实例创建所有数据库表
        db.create_all(app=app)
        # 打印数据库创建完成的提示信息
        print('Created Database!')

# 定义创建Flask应用的函数
def create_app():
    # 使用当前模块名称创建Flask应用实例
    app = Flask(__name__)
    # 设置应用密钥
    app.config['SECRET_KEY'] = #SECRET KEY#
    # 设置数据库URI
    app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{DB_NAME}'
    # 用Flask应用实例初始化SQLAlchemy对象
    db.init_app(app)

    # 导入'views'蓝图模块
    from .views import views
    # 导入'auth'蓝图模块
    from .auth import auth

    # 向应用实例注册'views'蓝图
    app.register_blueprint(views, url_prefix='/')
    # 向应用实例注册'auth'蓝图
    app.register_blueprint(auth, url_prefix='/')

    # 导入User模型
    from .models import User

    # 创建所有数据库表
    with app.app_context():
        db.create_all()

    # 创建LoginManager对象
    login_manager = LoginManager()
    # 设置应用的登录视图
    login_manager.login_view = 'auth.login'
    # 用Flask应用实例初始化LoginManager对象
    login_manager.init_app(app)

    # 定义根据ID从User模型加载用户的函数
    @login_manager.user_loader
    def load_user(id):
        return User.query.get(int(id))

    return app

一、解决循环导入问题 #

你当前在create_app内部延迟导入蓝图和模型的做法是正确的，只需确保其他模块也遵循相同规则：

• 禁止在模块顶部直接导入app，所有需要使用db或模型的逻辑，要么放在函数内部导入，要么从包内直接导入db/模型。
• 示例（views.py中）：

  from flask import Blueprint
  views = Blueprint('views', __name__)
  
  @views.route('/')
  def home():
      from .models import User
      # 在此处使用User进行数据库操作
      return "Home"
  

二、实现Twilio定时短信任务 #

Flask开发服务器是单线程的，直接在main.py或__init__.py中写定时任务会被阻塞，推荐用APScheduler实现后台定时任务，同时处理好Flask应用上下文：

步骤1：安装依赖 #

pip install apscheduler twilio

步骤2：新建定时任务模块 #

在website目录下创建scheduler.py：

from apscheduler.schedulers.background import BackgroundScheduler
from twilio.rest import Client
from .models import User
from . import db
from .keys import TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_PHONE_NUMBER

def send_daily_sms():
    # 必须在Flask应用上下文中执行数据库操作
    with db.app.app_context():
        # 获取所有订阅短信的用户
        subscribed_users = User.query.filter_by(gratitude_text_opt_in_status=True).all()
        client = Client(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)
        
        for user in subscribed_users:
            if user.phone_number:
                try:
                    client.messages.create(
                        body="每日感恩提醒：记得记录今天的小确幸哦！",
                        from_=TWILIO_PHONE_NUMBER,
                        to=user.phone_number
                    )
                except Exception as e:
                    print(f"发送短信给{user.phone_number}失败：{str(e)}")

def start_scheduler():
    scheduler = BackgroundScheduler()
    # 配置每日17点执行任务
    scheduler.add_job(send_daily_sms, 'cron', hour=17)
    scheduler.start()

步骤3：在应用初始化时启动任务 #

修改__init__.py的create_app函数，在末尾添加：

from .scheduler import start_scheduler

# ... 原有的create_app代码 ...

    # 启动后台定时任务
    start_scheduler()

    return app

步骤4：关键注意事项 #

• 禁止在debug=True模式下运行定时任务（debug模式会重启进程，导致任务重复启动），调试时可修改main.py：

  if __name__ == '__main__':
      app = create_app()
      import os
      # 仅在主进程启动定时任务
      if not app.debug or os.environ.get('WERKZEUG_RUN_MAIN') == 'true':
          from website.scheduler import start_scheduler
          start_scheduler()
      app.run(debug=True)
  

• 确保keys.py中正确配置Twilio密钥：

  TWILIO_ACCOUNT_SID = '你的Twilio账号SID'
  TWILIO_AUTH_TOKEN = '你的Twilio认证令牌'
  TWILIO_PHONE_NUMBER = '你的Twilio电话号码'
  

内容的提问来源于stack exchange，提问作者Doctors_Suck_at_Coding