Comment lier en toute sécurité la colonne Oracle à ORDER BY à SQLAlchemy dans une requête brute?

Question

J'essaie d'exécuter une requête sql brute et de passer en toute sécurité un ordre par/asc/desc en fonction de l'entrée de l'utilisateur. C'est l'extrémité arrière d'une grille de données paginée. Je ne peux pas pour la vie de moi comprendre comment faire cela en toute sécurité. Les paramètres sont convertis en chaînes afin qu'Oracle ne puisse pas exécuter la requête. Je ne trouve aucun exemple de ceci n'importe où sur Internet. Quelle est la meilleure façon d'accomplir cela en toute sécurité? (Je n'utilise pas l'ORM, doit être raw sql).

Ma solution de contournement consiste à définir ASC/DESC sur une variable que j'ai définie. Cela fonctionne bien et est sûr. Cependant, comment lier un nom de colonne à ORDER BY? Est-ce que c'est possible? Je peux juste ajouter à la liste blanche un groupe de colonnes et faire quelque chose de similaire à l'ASC/DESC. J'étais juste curieux de savoir s'il y avait un moyen de le lier. Merci.

@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: nom de la variable illégale/Numéro [SQL: « SELECT bb_request_id, bb_barcode, bs_status, TO_CHAR (bb_rec_cre_date, 'MM/JJ/AAAA) AS bb_rec_cre_date dE bars_barcodes, bars_status OU bs_status_id = bb_status_id ORDER BY: ord_by: ord_dir « ] [paramètres: { 'ord_by': u'bb_rec_cre_date », 'ord_dir': 'ASC'}]

Solution UPDATE basée sur acceptée answer:

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

Vous pouvez utiliser des constructions Core telles que table() et column() à la place des chaînes SQL brutes. Ce serait rendre votre vie plus facile à cet égard:

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() et column() représentent la partie syntaxique d'un objet complet soufflé Table avec Column s et peut être utilisé de cette façon à des fins qui s'échappent:

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.

Néanmoins, la liste blanche n'est peut-être pas une mauvaise idée.

Notez que vous n'avez pas besoin manuellement zip() les proxies de ligne afin de produire des dictionnaires. Ils agissent en tant que mappages tels quels, et si vous avez besoin de dict() pour des raisons de sérialisation ou autres, faites simplement dict(row).