从 python SQLAlchemy 中的部件构造 SQL Server 查询-解网

问：

Python = 3.10.2
SQLalchemy = 1.4.32
pyodbc = 4.0.32

是否可以从部分构建 sql 查询而不冒 SQL 注入的风险？这是我尝试过的此函数是已创建引擎的包装类的一部分

from sqlalchemy.sql import text
    def select_locations(self, country: str, include_neighbours: bool, neighbours: str):
    select_query = text("SELECT * FROM table_a")
    if include_neighbours is False:
        where_query = text("WHERE table_a.country = :country")
    else:
        where_query = text(
            "WHERE table_a.country = :country or table_a.country in (:neighbours)"
        )
    final_query = select_query + where_query

    if include_neighbours is False:
        params = {"country": country}
    else:
        params = {"country": country, "neighbours": neighbours}
    with self.engine.connect() as con:
        result = con.execute(final_query, params).fetchall()
    return result

但无法添加两个文本 clasues

TypeError: unsupported operand type(s) for +: 'TextClause' and 'TextClause'

最简单的方法是创建一个 python 字符串，但由于 SQL 注入，这是有风险的，所以它是不可能的。是创建原始 SQL 查询的可取方法。text

python sql sql-server sqlalchemy sql-injection

import sqlalchemy as sa

# …

table_a = sa.Table("table_a", sa.MetaData(), autoload_with=engine)

with engine.begin() as conn:
    # basic query
    qry = sa.select(sa.text("*")).select_from(table_a)
    results = conn.execute(qry).fetchall()
    """ SQL emitted:
    SELECT * 
    FROM table_a
    
    [generated in 0.00037s] ()
    """

    # add WHERE clause to basic query
    country = "Canada"
    qry = qry.where(table_a.c.country == country)
    results = conn.execute(qry).fetchall()
    """ SQL emitted:
    SELECT * 
    FROM table_a 
    WHERE table_a.country = ?

    [generated in 0.00033s] ('Canada',)
    """

上一个：如何使用带参数的表名的 Select 查询？[复制]

下一个：SonarQube 误报 SQL 注入

从 python SQLAlchemy 中的部件构造 SQL Server 查询

Constructing SQL Server query from parts in python SQLAlchemy

评论