如何在FastAPI中添加自定义验证？

Question

我正在构建一种 Youtube 音频下载器 api，我想验证视频 ID（使用 youtube_dl）。如何在 FastAPI 中添加自定义验证？

@router.get(
    "/audio/{video_id}",
    response_description="Redirects to the static url of the audio file.",
)
async def download_audio(
    video_id: str = Path(None, title="ID of the video (what you can see in the url)"),  # <-- How to validate?
):
...
# Here's some logic to download the audio of the video and save it. After that a `RedirectResponse` gets returned.

我知道我可以在函数中验证它，但我认为 FastAPI 有更好的选择。

Answer 1

FastAPI 使用pydantic进行数据验证，因此您可以使用标准库类型、Pydantic 类型或约束类型作为路径参数。

例子：

from fastapi import FastAPI
from pydantic import constr, NegativeInt


app = FastAPI(title="Test")


@app.get("/01/{test}")
async def test01(test: NegativeInt):
    return {"test": test}


@app.get("/02/{test}")
async def test02(test: constr(regex=r"^apple (pie|tart|sandwich)$")):
    return {"test": test}

测试：

$ curl -Ss localhost:8000/01/1 | python -m json.tool
{
    "detail": [
        {
            "loc": [
                "path",
                "test"
            ],
            "msg": "ensure this value is less than 0",
            "type": "value_error.number.not_lt",
            "ctx": {
                "limit_value": 0
            }
        }
    ]
}
$ curl -Ss localhost:8000/01/-1 | python -m json.tool
{
    "test": -1
}


$ curl -Ss localhost:8000/02/-1 | python -m json.tool
{
    "detail": [
        {
            "loc": [
                "path",
                "test"
            ],
            "msg": "string does not match regex \"^apple (pie|tart|sandwich)$\"",
            "type": "value_error.str.regex",
            "ctx": {
                "pattern": "^apple (pie|tart|sandwich)$"
            }
        }
    ]
}
$ curl -Ss localhost:8000/02/apple%20pie | python -m json.tool
{
    "test": "apple pie"
}

话虽如此，我想你最好的选择是正则表达式，例如constr(regex=r"^[0-9A-Za-z_-]{10}[048AEIMQUYcgkosw]$")，请参阅YouTube 视频 ID 格式了解详细信息。

2021 年 4 月 22 日星期四 22:16:14 UTC 2021 更新：

你可以尝试这样的事情（当然这只是一个例子）：

from __future__ import unicode_literals

import youtube_dl

from fastapi import FastAPI, HTTPException
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import FileResponse


URL = "https://www.youtube.com/watch?v="


app = FastAPI(title="Test")

ydl_opts = {
    "format": "bestaudio/best",
    "outtmpl": "%(id)s.%(ext)s",
    "postprocessors": [
        {
            "key": "FFmpegExtractAudio",
            "preferredcodec": "mp3",
            "preferredquality": "192",
        }
    ],
    "quiet": True,
}


def get_audio(video_id: str):
    with youtube_dl.YoutubeDL(ydl_opts) as ydl:
        try:
            yinfo = ydl.extract_info(f"{URL}{video_id}")
        except youtube_dl.DownloadError:
            ret = None
        else:
            ret = (f"{yinfo['title']}.mp3", f"{yinfo['id']}.mp3")

    return ret


@app.get("/audio/{video_id}")
async def download_audio(video_id: str):
    ret = await run_in_threadpool(get_audio, video_id)

    if not ret:
        raise HTTPException(status_code=418, detail="Download error or invalid ID")

    title, filename = ret

    return FileResponse(filename, filename=title)

测试：

$ time curl -Ss -D - -o rickroll.mp3 localhost:8000/audio/dQw4w9WgXcQ
HTTP/1.1 200 OK
date: Thu, 22 Apr 2021 22:26:50 GMT
server: uvicorn
content-type: audio/mpeg
content-disposition: attachment; filename*=utf-8''Rick%20Astley%20-%20Never%20Gonna%20Give%20You%20Up%20%28Video%29.mp3
content-length: 5090733
last-modified: Mon, 13 Jan 2020 17:04:18 GMT
etag: e26e47edb1401e6e65e4c8eb221f3419


real    0m11.883s
user    0m0.047s
sys 0m0.057s
$ file rickroll.mp3 
rickroll.mp3: Audio file with ID3 version 2.4.0, contains:MPEG ADTS, layer III, v1, 192 kbps, 48 kHz, Stereo