如何用Pydantic验证FastAPI音频上传的格式与时长？

问题分析 #

直接继承UploadFile和BaseModel无法生效，因为FastAPI的File(...)依赖会直接返回UploadFile实例，不会自动转换为自定义的Audio模型。另外，你之前的data验证逻辑错误——data是字节数据，直接和1比较毫无意义，需要解析音频文件获取实际时长。

解决方案 #

方案1：自定义依赖 + Pydantic验证 #

先定义验证规则，再通过自定义依赖处理文件并执行验证。如果需要验证音频时长，需安装音频处理库，示例使用pydub（支持多格式）：

pip install pydub python-multipart

（注：pydub依赖ffmpeg或libav，需提前在系统中安装）

代码实现：

from fastapi import FastAPI, File, UploadFile, Depends, HTTPException
from pydantic import BaseModel, validator
from pydub import AudioSegment
import io

app = FastAPI()

class AudioValidation(BaseModel):
    content_type: str
    duration: float

    @validator('content_type')
    def validate_content_type(cls, v):
        if not v.startswith("audio/"):
            raise HTTPException(status_code=400, detail="上传的不是音频文件")
        return v

    @validator('duration')
    def validate_duration(cls, v):
        if v <= 1.0:
            raise HTTPException(status_code=400, detail="音频时长必须超过1秒")
        return v

async def validate_audio_file(file: UploadFile = File(...)) -> UploadFile:
    # 先验证MIME类型
    AudioValidation(content_type=file.content_type, duration=0.0)
    
    # 读取文件并解析时长
    try:
        audio_data = await file.read()
        audio = AudioSegment.from_file(io.BytesIO(audio_data))
        duration = audio.duration_seconds
        # 验证时长
        AudioValidation(content_type=file.content_type, duration=duration)
        # 重置文件指针，避免后续读取失败
        await file.seek(0)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"无法解析音频文件：{str(e)}")
    
    return file

@app.post("/transcribe")
async def transcribe(
    lang: str = None, 
    file: UploadFile = Depends(validate_audio_file), 
    doctor_name: str = "Dr"
):
    audio = await file.read()
    # 后续转录逻辑
    return {"message": "验证通过", "filename": file.filename, "doctor_name": doctor_name}

方案2：使用Pydantic BeforeValidator（FastAPI 0.95+适用） #

直接对UploadFile字段绑定验证器，代码更简洁：

from fastapi import FastAPI, File, UploadFile, HTTPException
from pydantic import BeforeValidator, ValidationInfo
from typing import Annotated
from pydub import AudioSegment
import io

app = FastAPI()

def validate_audio(file: UploadFile, info: ValidationInfo) -> UploadFile:
    # 验证MIME类型
    if not file.content_type.startswith("audio/"):
        raise HTTPException(status_code=400, detail="上传的不是音频文件")
    
    # 验证时长
    try:
        audio_data = file.file.read()
        audio = AudioSegment.from_file(io.BytesIO(audio_data))
        duration = audio.duration_seconds
        if duration <= 1.0:
            raise HTTPException(status_code=400, detail="音频时长必须超过1秒")
        # 重置文件指针
        file.file.seek(0)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"无法解析音频文件：{str(e)}")
    
    return file

@app.post("/transcribe")
async def transcribe(
    lang: str = None, 
    file: Annotated[UploadFile, File(), BeforeValidator(validate_audio)], 
    doctor_name: str = "Dr"
):
    audio = await file.read()
    # 后续转录逻辑
    return {"message": "验证通过", "filename": file.filename}

关键说明 #

• 不要直接继承UploadFile和BaseModel，需通过自定义依赖或Pydantic验证器处理UploadFile实例。
• 音频时长必须解析文件内容判断，不能用字节大小替代——不同编码的音频，相同时长的字节数差异极大。
• 若仅需处理WAV格式，可使用Python标准库wave替代pydub，避免额外依赖：

import wave

def validate_wav_duration(audio_data: bytes):
    with wave.open(io.BytesIO(audio_data), 'rb') as wf:
        frames = wf.getnframes()
        rate = wf.getframerate()
        duration = frames / float(rate)
        return duration

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