FastAPI中自动删除过期OTP功能实现问题求助

问题排查点及修复方案 #

1. 异步函数混用同步SQLAlchemy操作 #

你的函数标记为async，但调用了同步的SQLAlchemy方法（db.query()、db.commit()），这会阻塞FastAPI的事件循环，导致任务无法正常执行。

• 若使用异步SQLAlchemy，需改用异步查询接口：

  from sqlalchemy import select
  
  # 异步查询示例
  expired_otp = await db.execute(select(OTPModel).filter(OTPModel.expires_at < datetime.utcnow().replace(tzinfo=pytz.utc)))
  expired_otp = expired_otp.scalars().all()
  

• 若使用同步SQLAlchemy，建议用asyncio.to_thread()包装同步操作，避免阻塞事件循环：

  from asyncio import to_thread
  
  async def delete_expired_otp():
      def sync_delete():
          db = next(get_db())
          # 同步删除逻辑
          db.commit()
          db.close()
      await to_thread(sync_delete)
  

2. 依赖注入在定时任务中失效 #

FastAPI的db_dependency基于请求上下文，定时任务运行时无请求上下文，无法自动注入数据库会话。需手动通过会话生成器（比如你定义的get_db()）获取会话。

3. 时区不匹配导致过滤条件失效 #

如果OTPModel.expires_at字段是带时区的（用DateTime(timezone=True)定义），而datetime.utcnow()生成的是无时区的naive时间，两者比较会出错。需转换为带时区的时间：

from datetime import datetime
import pytz

current_time = datetime.utcnow().replace(tzinfo=pytz.utc)
expired_otp = db.query(OTPModel).filter(OTPModel.expires_at < current_time).all()

4. 批量删除优化 #

无需循环删除每个OTP后都执行commit()，可批量删除后一次性提交，提升效率：

# 同步版本批量删除
db.query(OTPModel).filter(OTPModel.expires_at < current_time).delete(synchronize_session=False)
db.commit()

5. 验证repeat_every装饰器配置 #

确保使用的repeat_every是正确实现的，比如来自fastapi_utils.tasks的官方版本：

from fastapi_utils.tasks import repeat_every

自定义装饰器需确保能正确处理异步函数并按时触发。

完整修复示例代码 #

from datetime import datetime
import pytz
from asyncio import to_thread
from fastapi import FastAPI
from fastapi_utils.tasks import repeat_every
from sqlalchemy.orm import Session
from your_module import OTPModel, get_db

app = FastAPI()

@app.on_event('startup')
@repeat_every(seconds=300)
async def delete_expired_otp():
    print("deleting")
    
    def sync_delete_task():
        db: Session = next(get_db())
        try:
            current_time = datetime.utcnow().replace(tzinfo=pytz.utc)
            # 删除过期OTP
            db.query(OTPModel).filter(OTPModel.expires_at < current_time).delete(synchronize_session=False)
            # 或删除已使用的OTP
            # db.query(OTPModel).filter(OTPModel.is_used == True).delete(synchronize_session=False)
            db.commit()
        except Exception as e:
            db.rollback()
            print(f"删除失败: {str(e)}")
        finally:
            db.close()
    
    await to_thread(sync_delete_task)
    print("deleted")

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