如何使用 FastAPI 发布数据后下载文件？-IT科技

摘要：问题描述：我正在创建一个 Web 应用程序，它接收一些文本、将文本转换为语音，并返回一个 mp3 文件，该文件保存到临时目录中。我希望能够从 html 页面（即前端）下载文件，但我不知道如何正确地做到这一点。我知道使用 Flask 你可以做到这一点：from app import app from flask...

问题描述：

我正在创建一个 Web 应用程序，它接收一些文本、将文本转换为语音，并返回一个 mp3 文件，该文件保存到临时目录中。

我希望能够从 html 页面（即前端）下载文件，但我不知道如何正确地做到这一点。

我知道使用 Flask 你可以做到这一点：

from app import app
from flask import Flask, send_file, render_template
    
@app.route('/')
def upload_form():
    return render_template('download.html')

@app.route('/download')
def download_file():
    path = "html2pdf.pdf"

    return send_file(path, as_attachment=True)

if __name__ == "__main__":
    app.run()

HTML 示例：

<!doctype html>
<title>Python Flask File Download Example</title>
<h2>Download a file</h2>
<p><a href="{{ url_for('.download_file') }}">Download</a></p>

那么如何使用 FastAPI 复制这一点？

FastAPI代码：

from fastapi import FastAPI, File, Request, Response, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse
from fastapi.templating import Jinja2Templates
from gtts import gTTS

templates = Jinja2Templates(directory="templates")


def text_to_speech(language:str, text: str) -> str:
    tts = gTTS(text=text, lang=language, slow=False)
    tts.save("./temp/welcome.mp3")
    #os.system("mpg321 /temp/welcome.mp3")
    return "Text to speech conversion successful"


@app.get("/")
def home(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

@app.get("/text2speech")
async def home(request: Request):
    if request.method == "POST":
        form = await request.form()
        if form["message"] and form["language"]:
            language = form["language"]
            text = form["message"]
            translate = text_to_speech(language, text)
            path = './temp/welcome.mp3'
            value = FileResponse("./temp/welcome.mp3", media_type="audio/mp3")
            return value
    # return templates.TemplateResponse(
    #     "index.html",
    #     {"request": request, "message": text, "language": language, "download": value},
    # )

示例 HTML 文件：

<!doctype html>
<title>Download MP3 File</title>
<h2>Download a file</h2>
<p><a href="{{ url_for('text2speech') }}">Download</a></p>

解决方案 1：

使用Form关键字在您的端点中定义Form-data，更具体地说，使用Form(...)使参数成为必需，而不是使用await request.form()并手动检查用户是否提交了所需的参数。处理接收到的数据并生成音频文件后，您可以使用FileResponse将文件返回给用户。注意：使用headers中的参数FileResponse设置Content-Disposition标题，使用attachment参数（如此答案中所述）将文件下载到您的设备。未能设置headers或使用inline参数 isntead 会导致405 Method Not Allowed错误，因为在这种情况下，浏览器将尝试使用请求访问文件GET，以便以内联方式显示文件内容（但是，只POST允许向您的端点发出请求/text2speech）。请查看以下示例中的选项 1 。

如果您希望/text2speech端点同时支持GET和POST请求（如您的问题所示），您可以（1）使用@app.api_route("/text2speech", methods=["GET", "POST"])和使用request.method来检查它是否是GET或POST请求（请参阅此答案以获取工作示例），或（2）定义两个不同的端点，每个端点上使用以下装饰器，即和@app.post('/text2speech')（@app.get('/text2speech')旁注：这些装饰器也可以在同一个端点/函数上使用，然后request.method按照（1）选项中的说明使用）。但是，在这种情况下您不一定需要这样做。

此外，您还在模板中添加了Download超链接，供用户下载文件。但是，您尚未提供任何有关其如何运作的信息。如果您没有static文件，但有动态生成的音频文件（如您的情况），以及多个用户同时访问 API，那么这将无法运作；除非，例如，您为文件名生成了随机 UUID 并将文件保存到目录StaticFiles— 或者将该唯一标识符作为查询/路径参数（您也可以改用 cookie，请参阅此处和此处）添加到 URL 以识别要下载的文件 — 并将 URL 发送回用户。在这种情况下，您需要一个 Javascript 接口/库（如Fetch API ）来发出异步HTTP请求（如此答案中所述），以获取文件的 URL 并将其显示在Download超链接中。请查看下面的选项 2。注意：出于演示目的，选项 2 中的示例使用简单的方法dict将文件路径映射到 UUID。但是，在实际情况下，多个用户访问 API 并且可能使用多个工作人员，您可以考虑使用数据库存储或键值存储（缓存），如此处和此处所述。您还需要有一种机制，用于在下载文件后从数据库和磁盘中删除文件，并确保用户无权未经授权访问其他用户的音频文件。还值得一提的是，在同一个示例中，UUID 应作为端点的查询参数/download，但在实际情况下，您绝不应将敏感信息传递给查询字符串，因为这会带来安全/隐私风险（更多详细信息请参阅此答案的解决方案 1 ）。相反，您应该将敏感数据传递给请求正文（类似于此答案的解决方案 2 ），并始终使用 HTTPS 协议。

选项 1

应用程序

from fastapi import FastAPI, Request, Form
from fastapi.templating import Jinja2Templates
from fastapi.responses import FileResponse
import os

app = FastAPI()
templates = Jinja2Templates(directory="templates")


@app.get('/')
async def main(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})


@app.post('/text2speech')
def convert(request: Request, message: str = Form(...), language: str = Form(...)):
    # do some processing here
    filepath = './temp/welcome.mp3'
    filename = os.path.basename(filepath)
    headers = {'Content-Disposition': f'attachment; filename="{filename}"'}
    return FileResponse(filepath, headers=headers, media_type="audio/mp3")

上述方法的替代方法是读取端点内的文件数据（或者，如果数据事先已完全加载到内存中，例如此处、此处和此处）并直接返回自定义Response，如下所示：

from fastapi import Response

@app.post('/text2speech')
   ...
    with open(filepath, "rb") as f:
        contents = f.read()  # file contents could be already fully loaded into RAM
    
    headers = {'Content-Disposition': f'attachment; filename="{filename}"'}
    return Response(contents, headers=headers, media_type='audio/mp3')

如果您必须返回一个太大而无法放入内存的文件（例如，如果您有 8GB 的 RAM，则无法加载 50GB 的文件），则可以使用StreamingResponse，它会将文件分块加载到内存中，然后一次处理一个块的数据。如果您发现（yield from f如以下示例中所示）相当慢，请查看此答案以了解更快的替代方案。还应注意，使用FileResponse还会将文件内容分块加载到内存中（而不是一次加载全部内容）；但是，在这种情况下，块大小将是 64KB，如的实现类FileResponse中所指定。因此，如果该块大小不符合您的要求，您可以改用，如下所示，或者如本答案StreamingResponse所示，通过根据需要指定块大小。

from fastapi.responses import StreamingResponse

@app.post('/text2speech')
    ...
    def iterfile():
        with open(filepath, "rb") as f:
            yield from f

    headers = {'Content-Disposition': f'attachment; filename="{filename}"'}
    return StreamingResponse(iterfile(), headers=headers, media_type="audio/mp3")

模板/index.html

<!DOCTYPE html>
<html>
   <head>
      <title>Convert Text to Speech</title>
   </head>
   <body>
      <form method="post" action="http://127.0.0.1:8000/text2speech">
         message : <input type="text" name="message" value="This is a sample message"><br>
         language : <input type="text" name="language" value="en"><br>
         <input type="submit" value="submit">
      </form>
   </body>
</html>

使用 JavaScript 下载文件

如果你在前端使用 JavaScript 接口（例如 Fetch API）发出文件下载请求（而不是使用 HTML <form>，如上所示），请查看这个答案，以及这个答案和这个关于如何通过 JavaScript 在前端下载文件的答案。

选项 2

应用程序

from fastapi import FastAPI, Request, Form
from fastapi.templating import Jinja2Templates
from fastapi.responses import FileResponse
import uuid
import os

app = FastAPI()
templates = Jinja2Templates(directory="templates")
files = {}


@app.get('/')
async def main(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})


@app.get('/download')
def download_file(request: Request, fileId: str):
    filepath = files.get(fileId)
    if filepath:
        filename = os.path.basename(filepath)
        headers = {'Content-Disposition': f'attachment; filename="{filename}"'}
        return FileResponse(filepath, headers=headers, media_type='audio/mp3')    


@app.post('/text2speech')
def convert(request: Request, message: str = Form(...), language: str = Form(...)):
    # do some processing here
    filepath = './temp/welcome.mp3'
    file_id = str(uuid.uuid4())
    files[file_id] = filepath
    file_url = f'/download?fileId={file_id}'
    return {"fileURL": file_url}

模板/index.html

<!DOCTYPE html>
<html>
   <head>
      <title>Convert Text to Speech</title>
   </head>
   <body>
      <form method="post" id="myForm">
         message : <input type="text" name="message" value="This is a sample message"><br>
         language : <input type="text" name="language" value="en"><br>
         <input type="button" value="Submit" onclick="submitForm()">
      </form>

      <a id="downloadLink" href=""></a>

      <script type="text/javascript">
         function submitForm() {
             var formElement = document.getElementById('myForm');
             var data = new FormData(formElement);
             fetch('/text2speech', {
                   method: 'POST',
                   body: data,
                 })
                 .then(response => response.json())
                 .then(data => {
                   document.getElementById("downloadLink").href = data.fileURL;
                   document.getElementById("downloadLink").innerHTML = "Download";
                 })
                 .catch(error => {
                   console.error(error);
                 });
         }
      </script>
   </body>
</html>

下载后删除文件

对于上面的选项 1，要在用户下载文件后将其删除，您只需定义在BackgroundTask返回响应后运行的程序即可。例如：

from fastapi import BackgroundTasks
import os

@app.post('/text2speech')
def convert(request: Request, background_tasks: BackgroundTasks, ...):
    filepath = 'welcome.mp3'
    # ...
    background_tasks.add_task(os.remove, path=filepath)
    return FileResponse(filepath, headers=headers, media_type="audio/mp3")

但是，对于选项 2，您还必须确保file_id从缓存中删除指向给定文件路径的键（即）。因此，您应该创建一个后台任务函数，如下所示：

from fastapi import BackgroundTasks
import os

files = {}


def remove_file(filepath, fileId):
    os.remove(filepath)
    del files[fileId]


@app.get('/download')
def download_file(request: Request, fileId: str, background_tasks: BackgroundTasks):
    filepath = files.get(fileId)
    if filepath:
        # ...
        background_tasks.add_task(remove_file, filepath=filepath, fileId=fileId)
        return FileResponse(filepath, headers=headers, media_type='audio/mp3')

关于后台任务的更多详细信息和示例可以在这里以及这里找到。