如何使用 FastAPI 从内存缓冲区生成和返回 PDF 文件？

Question

我使用

PyFPDF

库生成一个 PDF 文件，并希望返回这个内存缓冲区而不将其写入本地磁盘。

@app.post("/pdf")
def generate_report(id_worker: int = Form(...),  start_date: date = Form(...), end_date: date = Form(...)):
    user = [id_worker]
    start_date = datetime(start_date.year, start_date.month, start_date.day)
    end_date = datetime(end_date.year, end_date.month, end_date.day)
    attendance = filter_by_date(zk, user, start_date, end_date)
    user_history = attendance_to_dict(attendance)
    dates, data, days, errors, updated_history = data_to_july(user_history, start_date, end_date)
    pdf = create_user_pdf(updated_history, start_date, end_date, days, errors)
    pdf_temp = "attendance.pdf"
    pdf.output(pdf_temp)
    name = "report.pdf"

    return FileResponse(pdf_temp, media_type="application/pdf", filename=name)

我试过使用：

name = "report.pdf"
pdf_buff = io.BytesIO()
pdf.output(pdf_buff)
pdf_buff.seek(0)
return FileResponse(pdf_buff.getvalue(), media_type="application/pdf", filename=name)

没有成功。有没有办法做到这一点？

Answer 1

我让它工作了。

@app.post("/pdf")
def generate_report(id_worker: int = Form(...),  start_date: date = Form(...), end_date: date = Form(...)):
    user = [id_worker]
    start_date = datetime(start_date.year, start_date.month, start_date.day)
    end_date = datetime(end_date.year, end_date.month, end_date.day)
    attendance = filter_by_date(zk, user, start_date, end_date)
    user_history = attendance_to_dict(attendance)
    dates, data, days, errors, updated_history = data_to_july(user_history, start_date, end_date)
    pdf = create_user_pdf(updated_history, start_date, end_date, days, errors)
    pdf_string = pdf.output(dest='S').encode('latin-1')
    pdf_buff = io.BytesIO(pdf_string)
    pdf_bytes = pdf_buff.getvalue()
    headers = {'Content-Disposition': 'attachment; filename="out.pdf"'}
    return Response(pdf_bytes, headers=headers, media_type='application/pdf')

我使用了@Chris的建议fastapi from buffer，fpdf to bytes和encoding fpdf使一切正常

Answer 2

我不建议使用

PyFPDF

，因为它已经过时并且不再被维护。相反，您可以使用

fpdf2

（也请参阅其 documentation），它是

PyFPDF

的分支和继承者，具有相似的语法。您可以按如下方式安装它：

pip install fpdf2

如 tutorial 中所述，当调用

FPDF.output()

方法而不提供任何文件路径参数时，该函数返回 PDF

bytearray

缓冲区，您可以使用

bytes

函数将其转换为

bytes()

对象并将其传递给custom

Response

（如果您需要将其作为 Jinja2 模板的一部分返回，请参见here）。如this answer中所述，要在浏览器中viewed PDF文件，您可以按如下方式设置

Content-Disposition

响应标头：

headers = {'Content-Disposition': 'inline; filename="out.pdf"'}

或者，要在浏览器中下载而不是查看 PDF 文件，请使用：

headers = {'Content-Disposition': 'attachment; filename="out.pdf"'}

这里有两个关于如何从 FastAPI 端点生成和返回 PDF 文件的选项。第一个选项使用用正常

def

定义的端点，而第二个选项使用

async def

端点。请查看这个答案以了解两者之间的区别，以及与

def

端点相比，FastAPI如何处理对

async def

端点的请求。

选项 1 - 使用

def

端点

from fastapi import FastAPI, Response
from fpdf import FPDF

app = FastAPI()


def create_PDF(text):
    pdf = FPDF()
    pdf.add_page()
    pdf.set_font('helvetica', 'B', 16)
    pdf.cell(10, 30, text)
    return pdf.output()

    
@app.get('/')
def get_pdf():
    out = create_PDF('Hello World!')
    headers = {'Content-Disposition': 'inline; filename="out.pdf"'}
    return Response(bytes(out), headers=headers, media_type='application/pdf')

选项 2 - 使用

async def

端点

from fastapi import FastAPI, Response
from fpdf import FPDF
import asyncio
import concurrent.futures

app = FastAPI()


def create_PDF(text):
    pdf = FPDF()
    pdf.add_page()
    pdf.set_font('helvetica', 'B', 16)
    pdf.cell(10, 30, text)
    return pdf.output()
    

@app.get('/')
async def get_pdf():
    loop = asyncio.get_running_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        out = await loop.run_in_executor(pool, create_PDF, 'Hello World!')
        
    headers = {'Content-Disposition': 'inline; filename="out.pdf"'}
    return Response(bytes(out), headers=headers, media_type='application/pdf')