解决Flask与Celery集成中的循环导入及任务未找到问题

问题根源 #

循环导入由app.py与tasks.py互相导入导致：tasks.py导入app.py中的业务函数，app.py顶部导入tasks.py中的process_ttml任务，形成双向依赖；任务未找到则是因为导入时机错误或依赖链断裂导致任务未被正确识别。

解决方案：解耦依赖，重构模块结构 #

1. 拆分业务逻辑到独立工具模块 #

新建utils.py，将app.py中被tasks.py依赖的业务函数（extract_transcript、summarize_transcript）移至该文件：

# utils.py
def extract_transcript(file_path):
    # 原业务实现代码
    pass

def summarize_transcript(transcript):
    # 原业务实现代码
    pass

2. 独立初始化Celery实例 #

修改celery_config.py，负责Celery实例的创建与配置，不依赖Flask应用：

# celery_config.py
from celery import Celery

def make_celery():
    celery = Celery(
        'tasks',
        broker='redis://localhost:6379/0',  # 你的Redis地址
        backend='redis://localhost:6379/0'
    )
    celery.conf.update(
        task_serializer='json',
        accept_content=['json'],
        result_serializer='json',
        timezone='UTC',
        enable_utc=True,
    )
    return celery

celery = make_celery()

3. 修改任务模块，依赖工具模块而非Flask应用 #

更新tasks.py，导入Celery实例与工具模块函数，注册任务：

# tasks.py
from celery_config import celery
from utils import extract_transcript, summarize_transcript

@celery.task
def process_ttml(file_path, include_timestamps):
    # 任务逻辑实现
    transcript = extract_transcript(file_path)
    summary = summarize_transcript(transcript)
    return {"transcript": transcript, "summary": summary}

4. 调整Flask应用的任务导入时机 #

在app.py中，不在顶部导入任务，而是在需要调用任务的函数内部延迟导入，同时改为依赖工具模块：

# app.py
from flask import Flask, request, jsonify
from utils import extract_transcript, summarize_transcript

app = Flask(__name__)

# 其他路由与初始化代码...

@app.route('/upload', methods=['POST'])
def upload_file():
    # 处理文件上传逻辑，获取file_path和include_timestamps
    # 延迟导入任务，避免循环依赖
    from tasks import process_ttml
    task = process_ttml.delay(file_path, include_timestamps)
    return jsonify({"task_id": task.id}), 202

if __name__ == '__main__':
    app.run(debug=True)

验证运行 #

1. 启动Celery Worker：

celery -A tasks.celery worker --loglevel=info

2. 启动Flask应用：

python app.py

核心原理 #

• 通过工具模块(utils.py)共享业务逻辑，消除app.py与tasks.py的直接双向依赖，形成单向依赖链：app.py→utils.py，tasks.py→utils.py。
• 延迟导入任务避免了模块加载时的循环冲突，同时确保Celery Worker能正确识别并加载任务。

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