生のクエリでOracle列をORDER BYにSQLAlchemyに安全にバインドするにはどうすればよいですか？

Question

私は未処理のSQLクエリを実行し、ユーザ入力に基づいて/ asc/descで安全にオーダーを渡そうとしています。これは、ページ付けされたDataGridのバックエンドです。私は私の人生のためにこれを安全に行う方法を理解できません。 Oracleが問合せを実行できないように、パラメータは文字列に変換されます。私はどこでもインターネット上のどこの例も見つけることができません。これを安全に達成する最良の方法は何ですか？ （私はORMを使用していない、生のSQLでなければならない）。

私の回避策は、私が設定した変数にASC/DESCを設定するだけです。これはうまく動作し、安全です。しかし、どのようにORDER BYに列名をバインドするのですか？それも可能ですか？列のホワイトリストを作成して、ASC/DESCと同じようなことをすることができます。それを縛る方法があれば私はちょうど興味があった。ありがとう。

@default.route('/api/barcodes/<sort_by>/<sort_dir>', methods=['GET']) 
@json_enc 
def fetch_barcodes(sort_by, sort_dir): 
    #time.sleep(5) 

    # Can't use sort_dir as a parameter, so assign to variable to sanitize it 
    ord_dir = "DESC" if sort_dir.lower() == 'desc' else 'ASC' 

    records = [] 
    stmt = text("SELECT bb_request_id,bb_barcode,bs_status, " 
     "TO_CHAR(bb_rec_cre_date, 'MM/DD/YYYY') AS bb_rec_cre_date " 
     "FROM bars_barcodes,bars_status " 
     "WHERE bs_status_id = bb_status_id " 
     "ORDER BY :ord_by :ord_dir ") 
    stmt = stmt.bindparams(ord_by=sort_by,ord_dir=ord_dir) 
    rs = db.session.execute(stmt) 
    records = [dict(zip(rs.keys(), row)) for row in rs] 

DatabaseError：（cx_Oracle.DatabaseError）ORA-01036：不正な変数名/番号 [SQL：「bb_request_idを選択、bb_barcode、bs_status、TO_CHAR（bb_rec_cre_date、 'MM/DD/YYYY'）AS bars_barcodes FROM bb_rec_cre_date、bars_status WHERE BY bs_status_id = bb_status_id ORDER：ord_by：ord_dir「] [パラメーター：{ 'ord_by'：u'bb_rec_cre_date」、 'ord_dir'： 'ASC'}受け入れに基づい]

UPDATEソリューション答え：

def fetch_barcodes(sort_by, sort_dir, page, rows_per_page): 
    ord_dir_func = desc if sort_dir.lower() == 'desc' else asc 
    query_limit = int(rows_per_page) 
    query_offset = (int(page) - 1) * query_limit 

    stmt = select([column('bb_request_id'), 
        column('bb_barcode'), 
        column('bs_status'), 
        func.to_char(column('bb_rec_cre_date'), 'MM/DD/YYYY').label('bb_rec_cre_date')]).\ 
     select_from(table('bars_barcode')).\ 
     select_from(table('bars_status')).\ 
     where(column('bs_status_id') == column('bb_status_id')).\ 
     order_by(ord_dir_func(column(sort_by))).\ 
     limit(query_limit).offset(query_offset) 

    result = db.session.execute(stmt) 
    records = [dict(row) for row in result] 
    response = json_return() 
    response.addRecords(records) 
    #response.setTotal(len(records)) 
    response.setTotal(1001) 
    response.setSuccess(True) 
    response.addMessage("Records retrieved successfully. Limit: " + str(query_limit) + ", Offset: " + str(query_offset) + " SQL: " + str(stmt)) 

    return response 

Answer 1

未加工のSQL文字列の代わりにtable()とcolumn()のようなCore構造を使用できます。それは、この点であなたの人生を容易にするだろう：

from sqlalchemy import select, table, column, asc, desc 

ord_dir = desc if sort_dir.lower() == 'desc' else asc 

stmt = select([column('bb_request_id'), 
       column('bb_barcode'), 
       column('bs_status'), 
       func.to_char(column('bb_rec_cre_date'), 
          'MM/DD/YYYY').label('bb_rec_cre_date')]).\ 
    select_from(table('bars_barcodes')).\ 
    select_from(table('bars_status')).\ 
    where(column('bs_status_id') == column('bb_status_id')).\ 
    order_by(ord_dir(column(sort_by))) 

table()とcolumn()はColumn sの本格Tableオブジェクトの構文部分を表すと目的を逃れるために、この方法で使用することができます。

The text handled by column() is assumed to be handled like the name of a database column; if the string contains mixed case, special characters, or matches a known reserved word on the target backend, the column expression will render using the quoting behavior determined by the backend.

まだ、ホワイトリストに登録するのは悪い考えではないかもしれません。

辞書を作成するために行のプロキシを手動でzip()する必要はありません。それらはそのままマッピングとして機能し、逐次化の理由などでdict()が必要な場合はdict(row)を実行します。