解耦Flask-SQLAlchemy：在应用外部执行数据库操作的教程

# app/database.py
from flask_sqlalchemy import SQLAlchemy

# 实例化 SQLAlchemy 对象，但暂不绑定到 Flask 应用
db = SQLAlchemy()

# app/models.py
import uuid
from sqlalchemy import func
# 从独立的 database 模块导入 db
from .database 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
    )

# app/app.py
from flask import Flask
# 从独立的 database 模块导入 db
from app.database import db
# 导入模型（如果需要，例如在 app.py 中定义路由或初始化数据时）
# from app.models import TokenBlocklist # 示例，实际可能不需要在这里直接导入

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

# 将 db 实例绑定到 Flask 应用
db.init_app(app)

# 在应用程序上下文中创建所有数据库表
with app.app_context():
    db.create_all()

# 其他 Flask 路由和应用逻辑
# ...

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

# scheduled_tasks/remove_old_tokens.py
from flask import Flask
from datetime import datetime, timedelta
import sys
import os

# 将项目根目录添加到 Python 路径，以便进行绝对导入
# 假设脚本在 app/scheduled_tasks/，项目根目录在 app/ 的父目录
sys.path.append(os.path.abspath('../../'))

# 从 app 模块中导入解耦的 db 实例和模型
from app.database import db
from app.models import TokenBlocklist


def remove_old_tokens():
    """
    删除创建时间超过40天的旧令牌。
    """
    forty_days = timedelta(days=40)
    forty_days_ago = datetime.now() - forty_days

    # 构建 SQLAlchemy 删除查询
    query = TokenBlocklist.__table__.delete().where(
        TokenBlocklist.created_at < forty_days_ago
    )

    # 执行查询并提交事务
    db.session.execute(query)
    db.session.commit()
    print('旧令牌已成功删除。')


# 为外部脚本创建一个最小的 Flask 应用实例
# 这允许 db 实例绑定到它，并提供应用程序上下文
app = Flask(__name__)

# 配置数据库 URI，确保路径正确
# 假设 db.sqlite 在 instance/ 目录下，相对于脚本的父目录的父目录
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' \
    + os.path.abspath('../../instance/db.sqlite')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# 将 db 实例绑定到这个临时的 Flask 应用
db.init_app(app)

# 在应用程序上下文中执行数据库操作
with app.app_context():
    # 确保数据库表存在（可选，如果 app.py 已经处理，这里可以省略）
    db.create_all()
    # 调用数据库操作函数
    remove_old_tokens()