如何通过sqlalchemy查询从Flask中流式传输CSV？

我想使用该技术they describe here从瓶流一个CSV：如何通过sqlalchemy查询从Flask中流式传输CSV？

from flask import Response 

@app.route('/large.csv') 
def generate_large_csv(): 
    def generate(): 
     for row in iter_all_rows(): 
      yield ','.join(row) + '\n' 
    return Response(generate(), mimetype='text/csv')

我从SQLAlchemy的查询返回购买对象的列表。我想将它写入CSV，理想情况是通过对输出到文件的属性进行自定义。

不幸的是，我得到一个空白CSV作为目前输出：

@app.route('/sales_export.csv') 
@login_required 
def sales_export(): 
    """ Export a CSV of all sales data """ 
    def generate(): 
     count = 0 
     fieldnames = [ 
      'uuid', 
      'recipient_name', 
      'recipient_email', 
      'shipping_street_address_1', 
      'shipping_street_address_2', 
      'shipping_city', 
      'shipping_state', 
      'shipping_zip', 
      'purchaser_name', 
      'purchaser_email', 
      'personal_message', 
      'sold_at' 
     ] 
     for i, row in enumerate(Purchase.query.all()): 
      if i == 0: 
       yield fieldnames 
      csv = ','.join(row) + '\n' 
      yield csv 
    return Response(generate(), mimetype='text/csv')

我要去哪里错了？

来源

2014-12-01 Patrick Beeson

您应该确定问题源于何处。可能是sqlalchmey没有返回数据，或者烧瓶没有正确处理它。尝试在控制台中打印Purchase.query.all的输出以检查sqlalchemy是否正在执行其应该的操作 – Marcom 2014-12-02 11:30:11

我的查询从控制台返回预期的对象列表。 – 2014-12-02 11:53:39

你有没有想过这个？ – diek 2015-01-24 18:02:42

下面是我如何解决这个问题。它不使用我最初对它感兴趣的流媒体。我仍然对使用该技术的解决方案感兴趣。

@app.route('/download') 
@login_required 
def download(): 
    "Export a CSV of all sales data" 
    purchases = Purchase.query.all() 
    csvfile = cStringIO.StringIO() 
    headers = [ 
     'uuid', 
     'recipient_name', 
     'recipient_email', 
     'shipping_street_address_1', 
     'shipping_street_address_2', 
     'shipping_city', 
     'shipping_state', 
     'shipping_zip', 
     'purchaser_name', 
     'purchaser_email', 
     'personal_message', 
     'sold_at', 
     'coupon_used' 
    ] 
    rows = [] 
    for purchase in Purchase.query.all(): 
     rows.append(
      { 
       'uuid': purchase.uuid, 
       'recipient_name': purchase.recipient_name, 
       'recipient_email': purchase.recipient_email, 
       'shipping_street_address_1': purchase.shipping_street_address_1, 
       'shipping_street_address_2': purchase.shipping_street_address_2, 
       'shipping_city': purchase.shipping_city, 
       'shipping_state': purchase.shipping_state, 
       'shipping_zip': purchase.shipping_zip, 
       'purchaser_name': purchase.purchaser_name, 
       'purchaser_email': purchase.purchaser_email, 
       'personal_message': purchase.personal_message, 
       'sold_at': purchase.sold_at.strftime('%c'), 
       'coupon_used': purchase.coupon_used 
      } 
     ) 
    writer = csv.DictWriter(csvfile, headers) 
    writer.writeheader() 
    for row in rows: 
     writer.writerow(
      dict(
       (k, v.encode('utf-8') if type(v) is unicode else v) for k, v in row.iteritems() 
      ) 
     ) 
    csvfile.seek(0) 
    return send_file(csvfile, attachment_filename='sales_export.csv', as_attachment=True)

来源

2015-01-26 16:57:41