输出CSV文件

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

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

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.body = output.getvalue()
class Report:

    async 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.body = output.getvalue()

我们在这里设置响应 Content-Type"text/csv" 根据推荐 RFC 4180 ,并指定可下载文件名 report.csv 通过 Content-Disposition 页眉(另请参见: 如何使用Falcon提供可下载的文件?

动态传输大型CSV文件

如果预计生成的CSV响应非常大,那么可能需要考虑动态流式CSV数据。这既可以避免存储整个响应数据的过多内存消耗,又可以减少查看者到第一个字节的时间。

为了动态地流化CSV行,我们将用我们自己的伪流对象初始化CSV writer,该对象实现 write() 方法,方法是在列表中累积数据。然后我们就出发 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()
class Report:

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

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

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

    async 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

            for chunk in stream.result:
                yield chunk
            stream.clear()

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

注解

在编写本文时,Python不支持 yield from 在异步生成器中,我们用循环表达式替换它。