输出CSV文件

生成CSV(或PDF等)报告并将其作为可下载文件提供是一项相当常见的后端服务任务。

最简单的方法是简单地将CSV行写入 io.StringIO 流,然后将其值赋给 resp.text

class Report:

    def on_get(self, req, resp):
        output = io.StringIO()
        writer = csv.writer(output, quoting=csv.QUOTE_NONNUMERIC)
        writer.writerow(('fruit', 'quantity'))
        writer.writerow(('apples', 13))
        writer.writerow(('oranges', 37))

        resp.content_type = 'text/csv'
        resp.downloadable_as = 'report.csv'
        resp.text = output.getvalue()

在这里,我们设置响应 Content-Type"text/csv" 按照以下组织的建议 RFC 4180 ,并指定可下载的文件名 report.csv 通过 Content-Disposition 标题(另请参阅: 如何使用Falcon提供可下载的文件? )。

动态传输大型CSV文件

如果预期生成的CSV响应非常大,则可能值得在生成CSV数据时将其流式传输。这种方法既可以避免过多的内存消耗,又可以减少查看器到达第一个字节的时间(TTFB)。

为了在飞翔上流式传输csv行,我们将使用我们自己的伪流对象初始化csv编写器。我们的小溪 write() 方法将简单地将CSV数据累加到列表中。然后我们将设置 resp.stream 提供给从该列表中生成数据块的生成器:

class Report:

    class PseudoTextStream:
        def __init__(self):
            self.clear()

        def clear(self):
            self.result = []

        def write(self, data):
            self.result.append(data.encode())

    def fibonacci_generator(self, n=1000):
        stream = self.PseudoTextStream()
        writer = csv.writer(stream, quoting=csv.QUOTE_NONNUMERIC)
        writer.writerow(('n', 'Fibonacci Fn'))

        previous = 1
        current = 0
        for i in range(n+1):
            writer.writerow((i, current))
            previous, current = current, current + previous

            yield from stream.result
            stream.clear()

    def on_get(self, req, resp):
        resp.content_type = 'text/csv'
        resp.downloadable_as = 'report.csv'
        resp.stream = self.fibonacci_generator()