如何通過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