FastAPI实现Project记录定时取消推荐功能方案咨询

针对需求，下面提供几种可行的实现方式，按场景区分说明：

一、FastAPI后台任务（快速实现，但局限性大） #

FastAPI自带的后台任务可以实现延迟执行，但注意：如果FastAPI进程重启，未完成的后台任务会直接丢失，仅适合测试环境或小型临时项目。

代码示例 #

1. 模型定义（补充必要字段）：

from sqlmodel import SQLModel, Field
from datetime import datetime

class Project(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    featured: bool = Field(default=False)
    # 可选：记录推荐开始时间，方便排查问题
    featured_at: datetime | None = Field(default=None)

2. 数据库依赖：

from sqlmodel import Session, create_engine

# 替换为你的实际数据库连接串
engine = create_engine("sqlite:///./test.db")  

def get_db():
    with Session(engine) as session:
        yield session

3. 后台任务与接口：

from fastapi import FastAPI, BackgroundTasks, Depends
from sqlmodel import Session
import time

app = FastAPI()

def auto_unfeature_project(project_id: int, db: Session):
    # 等待3天（转换为秒数）
    time.sleep(3 * 24 * 60 * 60)
    project = db.get(Project, project_id)
    if project and project.featured:
        project.featured = False
        db.commit()
        db.refresh(project)

@app.post("/projects/{project_id}/feature")
def set_project_featured(
    project_id: int,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    project = db.get(Project, project_id)
    if not project:
        return {"error": "Project not found"}
    
    project.featured = True
    project.featured_at = datetime.now()
    db.commit()
    db.refresh(project)
    
    # 添加后台任务，3天后自动取消推荐
    background_tasks.add_task(auto_unfeature_project, project_id, db)
    return {"message": "Project set as featured, will auto-unfeature in 3 days", "project": project}

二、数据库定时任务（可靠，适合轻量场景） #

通过数据库本身的定时调度功能实现，稳定性更高，不依赖FastAPI进程运行状态。

步骤 #

1. 扩展模型字段，记录推荐到期时间：

class Project(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    featured: bool = Field(default=False)
    featured_until: datetime | None = Field(default=None)

2. 设置推荐的接口：

from datetime import timedelta

@app.post("/projects/{project_id}/feature")
def set_project_featured(project_id: int, db: Session = Depends(get_db)):
    project = db.get(Project, project_id)
    if not project:
        return {"error": "Project not found"}
    
    project.featured = True
    # 设置3天后到期
    project.featured_until = datetime.now() + timedelta(days=3)
    db.commit()
    db.refresh(project)
    return {"message": "Project set as featured", "project": project}

3. 数据库层面添加定时任务：

• PostgreSQL（需先安装pg_cron扩展）：

-- 每天凌晨1点执行，取消已到期的推荐项目
SELECT cron.schedule(
    'auto-unfeature-projects',
    '0 1 * * *',
    'UPDATE project SET featured = false WHERE featured = true AND featured_until < NOW();'
);

• MySQL（需先开启事件调度器）：

SET GLOBAL event_scheduler = ON;

CREATE EVENT auto_unfeature_projects
ON SCHEDULE EVERY 1 DAY
STARTS '2024-01-01 01:00:00'
DO
UPDATE project SET featured = false WHERE featured = true AND featured_until < NOW();

三、Celery分布式任务队列（生产环境首选） #

如果是生产环境，需要可靠的任务调度、重试机制，推荐用Celery搭配Redis/RabbitMQ实现延迟任务。

代码示例 #

1. Celery配置：

from celery import Celery

# 替换为你的Redis/RabbitMQ地址
celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0"
)

2. 定义延迟任务：

from sqlmodel import Session, create_engine

# 替换为你的实际数据库连接串
engine = create_engine("你的数据库连接串")

@celery_app.task
def auto_unfeature_project(project_id: int):
    with Session(engine) as session:
        project = session.get(Project, project_id)
        if project and project.featured:
            project.featured = False
            session.commit()

3. FastAPI接口调用延迟任务：

from datetime import timedelta

@app.post("/projects/{project_id}/feature")
def set_project_featured(project_id: int, db: Session = Depends(get_db)):
    project = db.get(Project, project_id)
    if not project:
        return {"error": "Project not found"}
    
    project.featured = True
    db.commit()
    db.refresh(project)
    
    # 延迟3天执行取消推荐的任务
    auto_unfeature_project.apply_async(args=[project_id], countdown=3*24*60*60)
    return {"message": "Project set as featured, will auto-unfeature in 3 days", "project": project}

方案对比 #

• FastAPI后台任务：实现最快，但进程重启会丢失任务，仅适合测试/小项目。
• 数据库定时任务：无需额外服务，稳定性高，适合逻辑简单的场景。
• Celery：适合生产环境，支持分布式、任务重试、监控等功能，可靠性最强，但需要部署额外组件。

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