FastAPI中定期删除超期冻结账户的最佳实现方案咨询

结合你的代码结构和技术负责人的建议，在user.py中实现删除逻辑并通过if __name__ == "__main__"作为脚本执行是最符合现有架构的方案，以下是具体实现思路和细节：

方案优势 #

• 贴合现有职责划分：user.py负责账户数据库逻辑，将删除功能放在这里保持代码职责单一
• 复用现有MongoDB连接逻辑，无需重复编写数据库连接代码
• 兼顾两种调用场景：既可以被API通过依赖注入调用（如需手动触发删除），也可以作为独立脚本被cron定期执行
• 避免API调用的网络开销，直接操作数据库更高效

具体实现步骤 #

1. 在user.py中编写删除逻辑 #

利用base.py中已有的MongoDB连接逻辑，编写通用的删除函数，同时兼容依赖注入和脚本调用：

from datetime import datetime, timedelta
from base import get_db
import logging

# 配置日志（替代print，方便排查问题）
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def delete_suspended_accounts(db):
    # 计算7天前的UTC时间（避免时区偏差）
    cutoff_time = datetime.utcnow() - timedelta(days=7)
    # 执行删除：匹配冻结状态且冻结时间超过7天的账户
    delete_result = db.accounts.delete_many(
        {
            "status": "suspended",
            "suspended_at": {"$lte": cutoff_time}
        }
    )
    logger.info(f"成功删除 {delete_result.deleted_count} 个超期冻结账户")

# 脚本模式入口：直接调用函数执行删除
if __name__ == "__main__":
    # 直接获取数据库连接（无需依赖注入）
    db_instance = get_db()
    delete_suspended_accounts(db_instance)

2. 兼容FastAPI依赖注入调用（可选） #

如果需要在API中手动触发删除（比如后台管理接口），可以在authentication.py中导入该函数并通过依赖注入传参：

from fastapi import APIRouter, Depends
from user import delete_suspended_accounts
from base import get_db

router = APIRouter(prefix="/admin")

@router.post("/clean-suspended-accounts")
def clean_suspended_accounts(db=Depends(get_db)):
    delete_suspended_accounts(db)
    return {"message": "超期冻结账户清理完成"}

3. 配置cron定期执行 #

直接将user.py作为脚本加入cron任务，比如每天凌晨2点执行：

# 编辑cron任务
crontab -e
# 添加以下内容（替换为你的Python环境路径和项目路径）
0 2 * * * /path/to/your/python/env/bin/python /path/to/your/project/user.py

关键细节解答 #

• 关于复用base.py的MongoDB连接：无论哪种方案，都可以直接导入base.py中的连接或get_db函数，无需单独编写连接代码，上述方案已经实现了这一点
• FastAPI依赖注入的适配：脚本模式下不需要使用Depends，直接调用get_db()获取数据库实例即可，完美兼容现有连接逻辑

额外建议 #

• 添加执行锁机制（比如基于文件的锁），避免cron任务重复触发时出现并发执行问题
• 测试阶段先手动运行python user.py，验证删除逻辑的正确性后再配置cron
• 可以为删除操作添加备份逻辑（比如将删除的账户数据归档到历史集合），避免数据丢失

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