如何在原始查询中安全地将Oracle列绑定到ORDER BY到SQLAlchemy？

Question

我试图执行一个原始的sql查询，并根据用户输入安全地通过/ asc/desc传递一个命令。这是分页数据网格的后端。我不能为了我的生活而弄清楚如何安全地做到这一点。参数被转换为字符串，因此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：“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“] [参数：{ '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

您可以使用Core这样的结构，例如table()和column()来代替原始SQL字符串。这会让你的生活在这方面更容易：

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()代表一个完全成熟的Table对象的语法部分与Column S和可以以这种方式被用于转义用途：

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)。