FastAPI后台开发基础（19）：异步任务BackgroundTask

原创

bowenerchen

发布于 2024-11-20 10:20:49

21600

代码可运行

文章被收录于专栏：编码视界编码视界

运行总次数：0

代码可运行

在路径操作中使用BackgroundTask

def my_background_task(task_id: str):
    """
    测试 background_task
    """
    begin_time = time.time()
    time.sleep(random.randint(1, 5))
    end_time = time.time()
    print(f'task_id:{task_id}, task finished, time cost:{end_time - begin_time}s')


@app.post('/background_task')
async def background_task(background_tasks: BackgroundTasks):
    """
    测试接口 background-task
    """
    task_id = uuid.uuid4().hex
    background_tasks.add_task(my_background_task, task_id)
    return {
        'current_task_id': task_id
    }

结合依赖注入使用异步任务

def my_background_task(task_id: str):
    """
    测试 background_task
    """
    begin_time = time.time()
    time.sleep(random.randint(1, 5))
    end_time = time.time()
    print(f'task_id:{task_id}, task finished, time cost:{end_time - begin_time}s')

def get_q(background_tasks: BackgroundTasks, q: str | None = None):
    """
    测试接口 background-task
    """
    if q is not None:
        task_id = uuid.uuid4().hex
        q = f'{q}-{task_id}'
        background_tasks.add_task(my_background_task, task_id)
    return q


@app.post('/background_task_2')
async def background_task_2(q: Annotated[str, Depends(get_q)]):
    """
    测试接口 background-task
    """
    return {
        'current_task_id': q
    }