我正在try 编写响应以允许下载报告.我通过数据库查询检索相关数据,并将其存储在内存中,以避免在服务器上生成不必要的文件.我目前面临的挑战是将CSV文件保存到一个压缩文件中.遗憾的是,我在这个问题上花了几个小时也没有找到令人满意的解决方案,我也不确定我可能犯下的具体错误.有问题的CSV文件大小约为40 MB.

这是我的FastAPI代码.我成功地将CSV文件保存在本地,其中的所有数据都是准确的.我还设法正确地创建了一个包含CSV的压缩文件.然而,FastAPI响应的行为与预期不符.下载后,它返回我的压缩错误

ZIP文件已损坏,或者存档出现意外结尾.

from fastapi import APIRouter, Depends
from sqlalchemy import text
from libs.auth_common import veryfi_admin
from libs.database import database
import csv
import io
import zipfile
from fastapi.responses import Response

router = APIRouter(
    tags=['report'],
    responses={404: {'description': 'not found'}}
)


@router.get('/raport', dependencies=[Depends(veryfi_admin)])
async def get_raport():
    query = text(
        """
            some query
        """
    )

    data_de = await database.fetch_all(query)

    csv_buffer = io.StringIO()
    csv_writer_de = csv.writer(csv_buffer, delimiter=';', lineterminator='\n')

    csv_writer_de.writerow([
        "id", "name", "date", "stock",
    ])

    for row in data_de:
        csv_writer_de.writerow([
            row.id,
            row.name,
            row.date,
            row.stock,

        ])
    csv_buffer.seek(0)

    zip_buffer = io.BytesIO()
    with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
        zip_file.writestr("data.csv", csv_buffer.getvalue())

    response = Response(content=zip_buffer.getvalue())
    response.headers["Content-Disposition"] = "attachment; filename=data.zip"
    response.headers["Content-Type"] = "application/zip"
    response.headers["Content-Length"] = str(len(zip_buffer.getvalue()))

    print("CSV Buffer Contents:")
    print(csv_buffer.getvalue())
    return response

这也是VUE3代码

const downloadReport = () => {
  loading.value = true;
  instance
    .get(`/raport`)
    .then((res) => {
      const blob = new Blob([res.data], { type: "application/zip" });
      const link = document.createElement("a");
      link.href = window.URL.createObjectURL(blob);
      link.download = "raport.zip";
      link.click();
      loading.value = false;
    })
    .catch(() => (loading.value = false));
};
<button @click="downloadReport" :disabled="loading">
      Download Report
</button>

感谢您的理解,我在这个平台上提出了我的第一个问题.

推荐答案

下面是一个关于如何创建多个csv文件,然后将它们添加到zip文件,最后将zip文件返回给客户端的工作示例.此答案使用了前面在以下答案中讨论的代码和概念:thisthisthis.因此,我建议查看这些答案以了解更多细节.

另外,因为zipfile模块的操作是synchronous,所以您应该用普通的def而不是async def来定义端点,除非您使用了一些也提供asyncAPI的第三方库,或者您必须使用await来定义端点内的一些协程(async def函数),在这种情况下,我建议在外部ThreadPool中运行zipfile‘S操作(因为它们阻塞了IO绑定的操作).请查看this answer中的相关解决方案,以及有关async/await的详细信息,以及FastAPI如何处理async def和普通def API端点.

此外,如果数据已经加载到内存中,那么您实际上不需要使用StreamingResponse,如下面的示例所示.相反,您应该返回一个自定义的Response(参见下面的示例,以及thisthisthis以了解更多详细信息).

请注意,下面的示例对csv个数据使用utf-16编码,以使其与包括Unicode或非ASCII字符的数据兼容,如this answer中所述.如果您的数据中没有这样的字符,也可以使用utf-8编码.

另外,请注意,出于演示目的,下面的示例循环通过dict个对象中的list个对象来写入csv个数据,以便您更容易地使其适应您的数据库查询数据情况.否则,也可以使用csv.DictWriter()及其writerows()方法,如this answer中所示,以便写入数据,而不是循环通过list.

工作示例

from fastapi import FastAPI, HTTPException, BackgroundTasks, Response
import zipfile
import csv
import io


app = FastAPI()


fake_data = [
  {
    "Id": "1",
    "name": "Alice",
    "age": "20",
    "height": "62",
    "weight": "120.6"
  },
  {
    "Id": "2",
    "name": "Freddie",
    "age": "21",
    "height": "74",
    "weight": "190.6"
  }
]


def create_csv(data: list):
    s = io.StringIO()
    try:
        writer = csv.writer(s, delimiter='\t')
        writer.writerow(data[0].keys())
        for row in data:
            writer.writerow([row['Id'], row['name'], row['age'], row['height'], row['weight']])
        s.seek(0)
        return s.getvalue().encode('utf-16')
    except:
        raise HTTPException(detail='There was an error processing the data', status_code=400)
    finally:
        s.close()


@app.get('/')
def get_data():
    zip_buffer = io.BytesIO()
    try:
        with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
            for i in range(5):
                zip_info = zipfile.ZipInfo(f'data_{i}.csv')
                csv_data = create_csv(fake_data)
                zip_file.writestr(zip_info, csv_data)
            
        zip_buffer.seek(0)
        headers = {"Content-Disposition": "attachment; filename=files.zip"}
        return Response(zip_buffer.getvalue(), headers=headers, media_type="application/zip")
    except:
        raise HTTPException(detail='There was an error processing the data', status_code=400)
    finally:
        zip_buffer.close()

Python相关问答推荐

try 从网站获取表(ValueRight:如果使用所有纯量值,则必须传递索引)

在IIS中运行的FastAPI-获取权限错误:[Win错误10013]试图以其访问权限禁止的方式访问插槽

在pandas DataFrame上运行apply()时如何访问DateTime索引?

使用Python从HTTP打印值

Django关于UniqueBindition的更新

情节生成的饼图文本超出页面边界

想要使用Polars groupby_Dynamic来缩减时间序列收件箱(包括空垃圾箱)

Django序列化器没有验证或保存数据

将嵌套列表的字典转换为数据框中的行

用Python获取HTML Span类中的数据

Python上的Instagram API:缺少client_id参数"

删除所有列值,但判断是否存在任何二元组

NP.round解算数据后NP.unique

对象的`__call__`方法的setattr在Python中不起作用'

我想一列Panadas的Rashrame,这是一个URL,我保存为CSV,可以直接点击

将输入聚合到统一词典中

mypy无法推断类型参数.List和Iterable的区别

如何更新pandas DataFrame上列标题的de值?

python panda ExcelWriter切换动态公式到数组公式

在pandas/python中计数嵌套类别