我正在使用 SQLAlchemy 中的 Text 对象进行参数化查询,但出现错误。 工作示例:
import sqlalchemy as sqlal
from sqlalchemy.sql import text
db_table = 'Cars'
id_cars = 8
query = text("""SELECT *
FROM Cars
WHERE idCars = :p2
""")
self.engine.execute(query, {'p2': id_cars})
产生 sqlalchemy.exc.ProgrammingError 的示例:(pymysql.err.ProgrammingError) (1064,“您的 SQL 语法有错误)
import sqlalchemy as sqlal
from sqlalchemy.sql import text
db_table = 'Cars'
id_cars = 8
query = text("""SELECT *
FROM :p1
WHERE idCars = :p2
""")
self.engine.execute(query, {'p1': db_table, 'p2': id_cars})
如何使用受 SQL 注入保护的动态表名运行查询?
我使用 PostgreSQL 和 psycopg2 后端。我能够使用以下方法做到这一点:
from psycopg2 import sql
from sqlalchemy import engine
connection: sqlalchemy.engine.Connection
connection.connection.cursor().execute(
sql.SQL('SELECT * FROM {} where idCars = %s').format(sql.Identifier(db_table)),
(id_cars, )
)
PyMySQL 不支持。
如果您不是绝对需要使用完整文本查询,则可以使用 table 对象来定义 from 子句,并且 SQLAlchemy 将在需要时引用表名称。
where
子句(以及 order by
等)可以保留为文本,尽管它们必须放置在适当的方法调用中。
import sqlalchemy as sa
...
query = sa.select(sa.text('*')).select_from(sa.table(db_table)).where('idCars = :p2')
with engine.connect() as conn:
rows = conn.execute(query, {'p2': id_cars})
如果您能负担得起表反射的前期成本,您可以使用 Table 对象来代替,如果没有具有所提供名称的表,这将在反射时给您一个异常。
...
cars = sa.Table(db_table, sa.MetaData(), autoload_with=engine)
query = sa.select(cars).where(sa.text(idCars = :p2))
# or
query = sa.select(cars).filter_by(**{
...
一旦您拥有表实例,如果您想使用它,您就可以访问 SQLAlchemy(核心)API 的全部功能。
您可以利用用 python 编写的好处:
使用的库:
import sqlalchemy
from sqlalchemy import create_engine, MetaData, Table, func, event
from sqlalchemy.sql import text
from urllib.parse import quote_plus
连接(我在您的示例中没有看到 - 这里与 sql azure 连接):
params = urllib.parse.quote_plus(r'...')
conn_str = 'mssql+pyodbc:///?odbc_connect={}'.format(params)
engine_azure = create_engine(conn_str, echo=True)
你的例子:
db_table = 'Cars'
id_cars = 8
query = text('SELECT * FROM ' + db_table + 'WHERE idCars = ' + id_cars)
connection = engine.connect()
connection.execute(query)
connection.close()