Implosión de una lista para usar en una cláusula Python MySQL IN

7 minutos de lectura

avatar de usuario de mluebke
mluebke

Sé cómo asignar una lista a una cadena:

foostring = ",".join( map(str, list_of_ids) )

Y sé que puedo usar lo siguiente para convertir esa cadena en una cláusula IN:

cursor.execute("DELETE FROM foo.bar WHERE baz IN ('%s')" % (foostring))

Lo que necesito es lograr lo mismo de forma SEGURA (evitando la inyección de SQL) utilizando la base de datos MySQL. En el ejemplo anterior, debido a que no se pasa footstring como argumento para ejecutar, es vulnerable. También tengo que citar y escapar fuera de la biblioteca MySQL.

(Hay una pregunta SO relacionada, pero las respuestas enumeradas allí no funcionan para la base de datos MySQL o son vulnerables a la inyección de SQL).

  • Es posible que pueda inspirarse en una pregunta similar que se hace en php stackoverflow.com/questions/327274/…

    – Zoredache

    26 de febrero de 2009 a las 8:46

  • Posible duplicado de la lista de python en la consulta sql como parámetro

    – Kamil Sindi

    21 de febrero de 2016 a las 21:57

  • @mluebke ¿Alguna idea sobre cómo pasar varias listas en la consulta?

    – Dipen Dedania

    26 de agosto de 2016 a las 6:31

Utilizar el list_of_ids directamente:

format_strings=",".join(['%s'] * len(list_of_ids))
cursor.execute("DELETE FROM foo.bar WHERE baz IN (%s)" % format_strings,
                tuple(list_of_ids))

De esa manera, evita tener que citarse a sí mismo y evitar todo tipo de inyección de sql.

Tenga en cuenta que los datos (list_of_ids) va directamente al controlador de mysql, como un parámetro (no en el texto de la consulta), por lo que no hay inyección. Puede dejar los caracteres que desee en la cadena, sin necesidad de eliminar o citar caracteres.

  • @heikogerlach: No estoy citando el %s… La primera línea crea una cadena de “%s,%s,%s”… del mismo tamaño que la longitud de list_of_ids.

    – nosklo

    26 de febrero de 2009 a las 11:22

  • ¿Funcionará esto también en sqlite? Porque lo acabo de probar y parece señalar errores de sintaxis.

    – Sohaib

    16 de septiembre de 2014 a las 4:27


  • @Sohaib en sqlite el carácter de reemplazo es ? no %s por lo que funcionaría si cambia la primera línea a format_strings = ','.join('?' * len(list_of_ids)).

    – nosklo

    02/10/2014 a las 20:17

  • @kdas en tu caso no quieres el % format_strings parte para cambiar la otra %s marcadores de posición en su consulta, solo el IN (%s) marcador de posición: la forma de lograr esto es duplicar todos % caracteres excepto el que desea reemplazar: query = ("select distinct cln from vcf_commits where branch like %%s and repository like %%s and filename in (%s) and author not like %%s" % format_strings,); cursor.execute(query, (branch, repository) + tuple(fname_list) + (invalid_author,))

    – nosklo

    12 de marzo de 2019 a las 19:43

  • Aah, esto es brillante @nosklo. La coma (,) después de format_strings estaba causando un error, pero después de eliminarlo, funciona perfectamente. Genio. Aplausos. query = ("select distinct cln from vcf_commits where branch like %%s and repository like %%s and filename in (%s) and author not like %%s" % format_strings);

    – kdas

    14 de marzo de 2019 a las 5:42


Aunque esta pregunta es bastante antigua, pensé que sería mejor dejar una respuesta en caso de que alguien más estuviera buscando lo que yo quería.

La respuesta aceptada se complica cuando tenemos muchos parámetros o si queremos usar parámetros con nombre

Después de algunas pruebas

ids = [5, 3, ...]  # list of ids
cursor.execute('''
SELECT 
...
WHERE
  id IN %(ids)s
  AND created_at > %(start_dt)s
''', {
  'ids': tuple(ids), 'start_dt': '2019-10-31 00:00:00'
})

Probado con python2.7, pymysql==0.7.11

  • Esto no funciona con python 3 y mysql-connector-python 8.0.21. Se devuelve un error “La tupla de Python no se puede convertir al tipo MySQL”.

    – Rubros

    20 de julio de 2020 a las 6:40

Esto parece seguir siendo un problema con Python3 en 2021, como se señala en el comentario de Rubms a la respuesta de markk.

Agregar alrededor de 9 líneas de código al método “_process_params_dict” en “cursor.py” en el paquete del conector mysql para manejar tuplas me resolvió el problema:

def _process_params_dict(self, params):
    """Process query parameters given as dictionary"""
    try:
        to_mysql = self._connection.converter.to_mysql
        escape = self._connection.converter.escape
        quote = self._connection.converter.quote
        res = {}
        for key, value in list(params.items()):
            if type(value) is tuple: ### BEGIN MY ADDITIONS
                res[key.encode()] = b''
                for subvalue in value:
                    conv = subvalue
                    conv = to_mysql(conv)
                    conv = escape(conv)
                    conv = quote(conv)
                    res[key.encode()] = res[key.encode()] + b',' + conv if len(res[key.encode()]) else conv
            else: ### END MY ADDITIONS
                conv = value
                conv = to_mysql(conv)
                conv = escape(conv)
                conv = quote(conv)
                res[key.encode()] = conv
    except Exception as err:
        raise errors.ProgrammingError(
            "Failed processing pyformat-parameters; %s" % err)
    else:
        return res

Tal vez un poco tarde para la pregunta, pero me topé con un problema similar, pero quería usar un dict de parámetros con nombre en lugar de una tupla (porque si quiero modificar los parámetros para agregar o eliminar algunos, no quiero para reconstruir la tupla, estropear el orden puede ser muy fácil y provocar errores…).

Mi solución fue formatear la cadena de consulta para dividir el parámetro en varios parámetros y luego construir el dictado de parámetro con estos nuevos parámetros:

from typing import Iterable

query = """
SELECT *
FROM table
WHERE id IN (%(test_param)s)
"""

parameters = {"test_param": [1, 2, 3])

new_params = {}

for k, v in parameters.items():
    if isinstance(v, Iterable):
        iterable_params = {f"{k}_{i}": value for i, value in enumerate(v)}
        iterable_params_formatted = [f"%({k}_{i})s" for i in range(0, len(v))]
        query = query.replace(f"%({k})s", ", ".join(iterable_params_formatted))
        new_params.update(iterable_params)
    else:
        new_params[k] = v

print(query)
print(new_params)

Resultado:

> SELECT *
FROM table
WHERE id IN (%(test_param_0)s, %(test_param_1)s, %(test_param_2)s)

> {'test_param_0': 1, 'test_param_1': 2, 'test_param_2': 3}

Podría hacerse mejor, pero no pude encontrar una solución usando un dict de parámetros con nombre en lugar de una tupla ordenada.

Si utiliza Django 2.0 or 2.1 y Python 3.6Esta es la manera correcta:

from django.db import connection
RESULT_COLS = ['col1', 'col2', 'col3']
RESULT_COLS_STR = ', '.join(['a.'+'`'+i+'`' for i in RESULT_COLS])
QUERY_INDEX = RESULT_COLS[0]

TABLE_NAME = 'test'
search_value = ['ab', 'cd', 'ef']  # <-- a list
query = (
    f'SELECT DISTINCT {RESULT_COLS_STR} FROM {TABLE_NAME} a '
    f'WHERE a.`{RESULT_COLS[0]}` IN %s '
    f'ORDER BY a.`{RESULT_COLS[0]}`;'
)  # <- 'SELECT DISTINCT a.`col1`, a.`col2`, a.`col3` FROM test a WHERE a.`col1` IN %s ORDER BY a.`col1`;'
with connection.cursor() as cursor:
    cursor.execute(query, params=[search_value])  # params is a list with a list as its element

referencia: https://stackoverflow.com/a/23891759/2803344
https://docs.djangoproject.com/en/2.1/topics/db/sql/#passing-parameters-into-raw

avatar de usuario de prajmus
prajmus

list_of_ids = [ 1, 2, 3]
query = "select * from table where x in %s" % str(tuple(list_of_ids))
print query

Esto podría funcionar para algunos casos de uso si no desea preocuparse por el método en el que tiene que pasar argumentos para completar la cadena de consulta y le gustaría invocar solo cursror.execute(query).

Otra forma podría ser:

"select * from table where x in (%s)" % ', '.join(str(id) for id in list_of_ids)

avatar de usuario de chenchuk
chenchuk

Otra solución simple usando la comprensión de listas:

# creating a new list of strings and convert to tuple
sql_list = tuple([ key.encode("UTF-8") for key in list_of_ids ])

# replace "{}" with "('id1','id2',...'idlast')"
cursor.execute("DELETE FROM foo.bar WHERE baz IN {}".format(sql_list))

¿Ha sido útil esta solución?