FastAPI视频录制API优化咨询：无需数据库存储状态的实现方案

核心思路 #

放弃数据库，改用异步安全的内存存储管理每个相机的录制状态——用全局字典+异步锁，直接在内存中存储录制控制标志、后台任务对象、临时文件路径等信息，既简化架构又提升响应速度。

具体实现方案 #

1. 定义全局状态管理结构 #

用一个字典存储所有相机的录制状态，每个相机ID对应一个包含异步锁、控制标志、任务对象和临时文件名的字典，确保多请求并发操作时的线程安全。

2. 启动录制接口（/{camera}/start） #

• 检查目标相机是否处于录制状态，避免重复启动
• 创建临时MP4文件（可使用tempfile模块生成唯一文件名）
• 启动异步后台录制任务（用asyncio.create_task而非FastAPI自带的Background Task，因为后者仅在请求周期内运行）
• 更新全局状态，立即返回启动成功提示

3. 停止录制接口（/{camera}/stop） #

• 检查目标相机是否正在录制
• 设置控制标志为False，通知后台任务停止捕获
• 等待后台录制任务结束，确保文件写入完成
• 读取临时MP4文件，以流式方式返回给客户端
• 清理全局状态和临时文件

代码示例 #

from fastapi import FastAPI, HTTPException, Response
import asyncio
import tempfile
import os
from typing import Dict, Optional

app = FastAPI()

# 全局状态存储：key为相机ID，value为状态字典
camera_records: Dict[str, Dict] = {}

async def record_video(camera_id: str, temp_file_path: str):
    """后台录制任务：替换为实际视频捕获逻辑"""
    record_state = camera_records[camera_id]
    try:
        # 示例：用OpenCV读取相机流并写入文件的逻辑可放在这里
        with open(temp_file_path, 'wb') as f:
            while record_state['should_continue']:
                # 模拟写入视频帧数据，实际替换为相机流读取
                f.write(b'fake_video_frame_data')
                await asyncio.sleep(0.1)
    finally:
        # 确保即使出错也能重置控制标志
        record_state['should_continue'] = False

@app.get("/{camera_id}/start")
async def start_recording(camera_id: str):
    if camera_id in camera_records:
        raise HTTPException(status_code=400, detail=f"Camera {camera_id} is already recording")
    
    # 创建临时MP4文件
    temp_file = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False)
    temp_file.close()
    
    # 初始化相机状态
    camera_records[camera_id] = {
        'lock': asyncio.Lock(),
        'should_continue': True,
        'task': None,
        'temp_file_path': temp_file.name
    }
    
    record_state = camera_records[camera_id]
    async with record_state['lock']:
        # 启动异步录制任务
        record_state['task'] = asyncio.create_task(record_video(camera_id, temp_file.name))
    
    return {"message": f"video recording started for camera {camera_id}"}

@app.get("/{camera_id}/stop")
async def stop_recording(camera_id: str):
    if camera_id not in camera_records:
        raise HTTPException(status_code=400, detail=f"Camera {camera_id} is not recording")
    
    record_state = camera_records[camera_id]
    async with record_state['lock']:
        if not record_state['should_continue']:
            raise HTTPException(status_code=400, detail=f"Camera {camera_id} has already stopped recording")
        
        # 触发录制停止
        record_state['should_continue'] = False
        # 等待录制任务完成
        await record_state['task']
    
    # 返回MP4文件
    try:
        with open(record_state['temp_file_path'], 'rb') as f:
            video_data = f.read()
        return Response(content=video_data, media_type="video/mp4")
    finally:
        # 清理临时文件和全局状态
        os.unlink(record_state['temp_file_path'])
        del camera_records[camera_id]

关键注意事项 #

• 并发安全：每个相机状态绑定asyncio.Lock，避免多请求同时修改录制状态
• 临时文件管理：用tempfile生成唯一文件，停止录制后立即删除，避免磁盘冗余
• 任务可靠性：录制任务加入finally块，确保捕获过程出错时也能重置控制标志
• 重启风险：内存存储状态在服务重启后会丢失，若需持久化可替换为Redis等轻量存储，仍无需数据库
• 实际捕获逻辑：record_video函数为模拟代码，需替换为实际的相机流读取（如OpenCV）和视频编码逻辑

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