如何自动化传递给 pandas.read_sql 的参数？

Question

我正在尝试创建一种通过 locals() 之类的方式自动传递参数的方法，类似于 f 字符串的工作方式。

目前的工作方式

import pandas as pd

def my_func(conn, string_id, date, integer_ids):
    sql = f"""    
    select * from RISK a
    where STRING_ID = '{string_id}'
    and DATE = {date}
    and INTEGER_ID in ({','.join(map(str, integer_ids))})"""
    df = pd.read_sql(sql, conn)
    return df

但是，这种方法意味着我无法将 SQL 复制粘贴到 SQL developer 或类似软件中，然后从那里运行它。所以我想要一种使用参数的方法。

这似乎有两个问题

1. 参数必须是字面量，所以不能传递列表
2. 我需要手动创建一个字典，不能简单地传递locals()之类的东西

我希望它的工作方式类似于下面的示例（显然行不通）

import pandas as pd

def my_func(conn, string_id, date, integer_ids):
    sql = """    
    select * from RISK
    where STRING_ID = :string_id
    and DATE = :date
    and INTEGER_ID in :integer_ids"""
    df = pd.read_sql(sql, conn, params=locals())
    return df

编辑：经过一番测试，也许我可以使用正则表达式来查找:param 的所有实例并将它们替换为参数值，例如

import re
pattern = '[\n\r].*:\s*([^\n\r]*)'
matches = re.findall(pattern,sql)
for match in matches:
    sql = sql.replace(':'+match, eval(match))

它不是很漂亮，它引入了与对象类型相关的问题。例如。 string_id要封装成'，日期也要转成字符串对象

最终编辑：感谢 perl，我现在有一个可行的解决方案来解决我的问题

def read_sql(sql, conn, params):
    # Finds all words following a ":" sign in the sql
    for p in re.findall(':(\w+)', sql):
        if isinstance(params.get(p), (tuple, list)):
            ext_params = {f'{p}_{i:03d}': p_i for i, p_i in enumerate(params.get(p))}
            sql = sql.replace(f':{p}', f"(:{', :'.join(ext_params)})")
            params.update(ext_params)

    sql_text = sqlalchemy.text(sql)
    return pd.read_sql(sql_text, conn, params=params)


def my_func(conn, string_id, date, integer_ids):
    sql = """    
    select * from RISK
    where STRING_ID = :string_id
    and DATE = :date
    and INTEGER_ID in :integer_ids"""
    df = read_sql(sql, conn, locals())
    return df

Answer 1

您可以通过将查询包装在sqlalchemy.text 中并将列表转换为元组来使用参数化查询。例如：

def my_func(conn, min_number, letters):
    # convert lists to tuples
    letters = tuple(letters)
    
    # wrap sql in sqlalchemy.text
    sql = sqlalchemy.text("""    
        SELECT *
        FROM letters
        WHERE
            number >= :min_number AND
            letter in :letters""")
    
    # read and return the resulting dataframe
    df = pd.read_sql(sql, conn, params=locals())
    return df

my_func(conn, 10, ['a', 'b', 'c', 'x', 'y', 'z'])

输出：

  letter  number
0      x      23
1      y      24
2      z      25

---

为了示例的完整性，以下用作测试表：

df = pd.DataFrame({
    'letter': list(string.ascii_lowercase),
    'number': range(len(string.ascii_lowercase))})
df.to_sql('letters', conn, index=False)

---

更新：以下是 Oracle 使其适用于列表的可能解决方法：

def get_query(sql, **kwargs):
    for k, v in kwargs.items():
        vs = "','".join(v)
        sql = sql.replace(f':{k}', f"('{vs}')")
    return sql

def my_func(conn, min_number, letters):
    sql_template = """    
        SELECT *
        FROM letters
        WHERE
            number >= :min_number AND
            letter in :letters
    """
    # pass list variables to `get_query` function as named parameters
    # to get parameters replaced with ('value1', 'value2', ..., 'valueN')
    sql = sqlalchemy.text(
        get_query(sql_template, letters=letters))
    
    df = pd.read_sql(sql, conn, params=locals())
    return df

my_func(conn, 10, ['a', 'b', 'c', 'x', 'y', 'z'])

---

更新 2：这是适用于字符串和数字的 get_query 函数（用引号括起来的字符串，但不是数字）：

def get_query(sql, **kwargs):
    # enclose in quotes strings, but not numbers
    def q(x):
        q = '' if isinstance(x, (int, float)) else "'"
        return f'{q}{x}{q}'
    
    # replace with values
    for k, v in kwargs.items():
        sql = sql.replace(f':{k}', f"({','.join([q(x) for x in v])})")

    return sql

例如：

sql = """    
SELECT *
FROM letters
WHERE
    number in :numbers AND
    letters in :letters
"""

get_query(sql,
          numbers=[1, 2, 3],
          letters=['A', 'B', 'C'])

输出：

SELECT *
FROM letters
WHERE
    number in (1,2,3) AND
    letters in ('A','B','C')

Answer 2

为什么不这样：

import pandas as pd

def my_func(conn, string_id, date, integer_ids):
    sql = """    
    select * from RISK
    where STRING_ID = %s
    and DATE = %s
    and INTEGER_ID in %s"""
    df = pd.read_sql(sql, conn, (string_id, date,integer_ids))
    return df