输出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()
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.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()
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
在异步生成器中,我们用循环表达式替换它。