I have this piece of code that I’m using to do insert/update in Postgres, but I’m not sure if my construction of the SQL string literal is susceptible to SQL injection. If it is, how may I re-write the query?
<code>def build_sql_ident(cls, data) -> sql.Composed:
result = None
if isinstance(data, (list, tuple)):
result = sql.SQL(",").join([sql.Identifier(column) for column in data])
if isinstance(data, str):
result = sql.Identifier(data)
return result
tbl = build_sql_ident(d.get("tbl"))
cols = build_sql_ident(get_all_fields()) # get_all_fields() is a List[str]
q_id = build_sql_ident("q_id")
if subject:
clause = "SET (col4) = ROW(EXCLUDED.val4)"
else:
clause = "SET (col1, col2) = ROW(EXCLUDED.val1, EXCLUDED.val2)"
curs.execute(
sql.SQL(f"""INSERT INTO {{tbl}} ({{cols}}) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT ({{q_id}}) DO UPDATE {clause}""").format(tbl=tbl, cols=cols, q_id=q_id),
tuple(vals),
)
</code>
<code>def build_sql_ident(cls, data) -> sql.Composed:
result = None
if isinstance(data, (list, tuple)):
result = sql.SQL(",").join([sql.Identifier(column) for column in data])
if isinstance(data, str):
result = sql.Identifier(data)
return result
tbl = build_sql_ident(d.get("tbl"))
cols = build_sql_ident(get_all_fields()) # get_all_fields() is a List[str]
q_id = build_sql_ident("q_id")
if subject:
clause = "SET (col4) = ROW(EXCLUDED.val4)"
else:
clause = "SET (col1, col2) = ROW(EXCLUDED.val1, EXCLUDED.val2)"
curs.execute(
sql.SQL(f"""INSERT INTO {{tbl}} ({{cols}}) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT ({{q_id}}) DO UPDATE {clause}""").format(tbl=tbl, cols=cols, q_id=q_id),
tuple(vals),
)
</code>
def build_sql_ident(cls, data) -> sql.Composed:
result = None
if isinstance(data, (list, tuple)):
result = sql.SQL(",").join([sql.Identifier(column) for column in data])
if isinstance(data, str):
result = sql.Identifier(data)
return result
tbl = build_sql_ident(d.get("tbl"))
cols = build_sql_ident(get_all_fields()) # get_all_fields() is a List[str]
q_id = build_sql_ident("q_id")
if subject:
clause = "SET (col4) = ROW(EXCLUDED.val4)"
else:
clause = "SET (col1, col2) = ROW(EXCLUDED.val1, EXCLUDED.val2)"
curs.execute(
sql.SQL(f"""INSERT INTO {{tbl}} ({{cols}}) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT ({{q_id}}) DO UPDATE {clause}""").format(tbl=tbl, cols=cols, q_id=q_id),
tuple(vals),
)
1