如何在Python Django中实现多API并行调用并动态更新UI？

1. 异步并行调用API的正确实现 #

在Django中处理长时间外部API调用，推荐用**异步视图+aiohttp**实现，避免阻塞主线程：

步骤1：安装依赖 #

pip install aiohttp

步骤2：编写异步视图函数 #

# views.py
import asyncio
import aiohttp
from django.http import JsonResponse
from django.views import View
from django.utils.decorators import classonlymethod

class TaskProgressView(View):
    @classonlymethod
    def as_view(cls, **initkwargs):
        view = super().as_view(**initkwargs)
        view._is_coroutine = asyncio.coroutines._is_coroutine
        return view

    async def post(self, request):
        # 获取上传的文件
        uploaded_file = request.FILES.get('file')
        if not uploaded_file:
            return JsonResponse({'error': '文件不能为空'}, status=400)

        async with aiohttp.ClientSession() as session:
            # 构造文件上传参数
            form_data = aiohttp.FormData()
            form_data.add_field('file', uploaded_file.read(), filename=uploaded_file.name)
            
            # 异步启动计算API任务，不阻塞后续逻辑
            calc_task = asyncio.create_task(
                session.post('http://your-calc-api-url/', data=form_data)
            )

            progress = 0
            task_id = None
            while progress < 100:
                # 尝试获取计算API返回的task_id（若API需要用task_id查进度）
                if not task_id:
                    try:
                        calc_response = await asyncio.wait_for(calc_task, timeout=10)
                        task_id = (await calc_response.json()).get('task_id')
                    except asyncio.TimeoutError:
                        # 计算API未返回，继续执行进度查询
                        pass

                # 调用健康检查API更新进度
                if task_id:
                    progress_response = await session.get(f'http://your-health-api-url/?task_id={task_id}')
                    progress_data = await progress_response.json()
                    progress = progress_data.get('progress', 0)
                
                # 非阻塞等待10秒
                await asyncio.sleep(10)

            # 任务完成后获取最终结果
            final_result = await calc_task.json()
            return JsonResponse({'status': 'completed', 'result': final_result})

关键说明 #

• 用asyncio.create_task将计算API调用转为异步任务，与健康检查循环并行执行，避免阻塞Django事件循环
• 用aiohttp.ClientSession管理HTTP连接，比同步请求更高效
• 循环中用await asyncio.sleep(10)实现非阻塞等待，不会卡住服务

2. Django中动态更新前端进度的合理方案 #

前端不能直接等待后端长任务，推荐两种方案：

方案A：前端轮询（简单易实现） #

后端部分：新增进度查询接口

# views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt

# 生产环境建议用Redis/数据库存储进度，此处用内存字典临时演示
task_progress = {}

@csrf_exempt
def get_progress(request):
    if request.method == 'GET':
        task_id = request.GET.get('task_id')
        if not task_id:
            return JsonResponse({'error': 'task_id必填'}, status=400)
        progress = task_progress.get(task_id, 0)
        return JsonResponse({'progress': progress})

修改异步视图，将进度存入存储介质：

# 在健康检查循环中添加
if task_id:
    progress_response = await session.get(f'http://your-health-api-url/?task_id={task_id}')
    progress_data = await progress_response.json()
    progress = progress_data.get('progress', 0)
    task_progress[task_id] = progress

前端部分：JavaScript轮询更新UI

<!-- template.html -->
<div>
    <progress id="progressBar" value="0" max="100"></progress>
    <span id="progressText">0%</span>
</div>

<script>
function checkProgress(taskId) {
    fetch(`/get-progress/?task_id=${taskId}`)
        .then(response => response.json())
        .then(data => {
            const progress = data.progress;
            document.getElementById('progressBar').value = progress;
            document.getElementById('progressText').textContent = `${progress}%`;
            
            if (progress < 100) {
                setTimeout(() => checkProgress(taskId), 10000);
            } else {
                document.getElementById('progressText').textContent = '任务完成！';
            }
        })
        .catch(error => console.error('查询进度失败:', error));
}

// 提交文件后启动轮询
document.getElementById('fileForm').addEventListener('submit', async function(e) {
    e.preventDefault();
    const formData = new FormData(this);
    const response = await fetch('/start-task/', {
        method: 'POST',
        body: formData
    });
    const result = await response.json();
    if (result.task_id) {
        checkProgress(result.task_id);
    }
});

方案B：WebSocket实时推送（低延迟） #

如果需要更低延迟的更新，可使用Django Channels实现WebSocket：

1. 安装channels和channels-redis（生产环境依赖）
2. 配置WebSocket路由，后端获取进度后主动推送给前端
3. 前端通过WebSocket连接接收进度，实时刷新UI

生产环境注意事项 #

• 内存存储进度不可靠，必须用Redis或数据库持久化任务ID与进度
• 可结合Celery管理异步任务，但调用外部API用异步视图已足够
• 前端轮询需添加异常重试机制，避免网络波动导致进度中断

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