时间戳

首页 > 解决方案 > 在 Connection.execute sqlalchemy 中运行 UPDATE SET 参数化 sql 语句

python - 在 Connection.execute sqlalchemy 中运行 UPDATE SET 参数化 sql 语句

问题描述

我想在flask_sqlalchemy 中运行一个参数化的mysql UPDATE SET 语句。由于我不知道应该更新哪些列，我编写了一个辅助函数来帮助编写语句。

我的模型

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True, autoincrement= True)
    username = db.Column(db.String(80), nullable=False)
    email = db.Column(db.String(120), nullable=False)

    def __repr__(self):
        return '<User %r>' % self.username

辅助函数

def run_sql(params):
    # e.g: params = {'username': "testing", "email": "testing@testing.ts", "id": 1}
    id = params.pop("id")
    # params = {'username': "testing", "email": "testing@testing.ts"}
    sets = list(map(lambda col: f"{col}=:{col}", params.keys()))
    # sets = ["username=:username", "email=:email"]
    sets = ", ".join(sets)
    # sets = "username=:username, email=:email"
    params["id"] = id
    # params = {'username': "testing", "email": "testing@testing.ts", "id": 1}
    sql_statement = f"""UPDATE User SET {sets} WHERE id=:id LIMIT 1"""
    # sql_statement = UPDATE User SET username=:username, email=:email WHERE id=:id LIMIT 1
    return  sql_statement

调用辅助函数

if __name__ == "__main__":
    conn = engine.connect()
    params = {'username': "testing", "email": "testing@testing.ts", "id": 1}
    sql_statement = run_sql(params)
    conn.execute(sql_statement, params)

运行前面的代码会产生以下异常

"sqlalchemy.exc.ProgrammingError: (pymysql.err.ProgrammingError) (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near ':username, email=:email WHERE id=:id LIMIT 1' at line 1") [SQL: 'UPDATE User SET username=:username, email=:email WHERE id=:id LIMIT 1'] [parameters: {'username': 'testing', 'email': 'testing@testing.ts', 'id': 1}] (Background on this error at: http://sqlalche.me/e/f405)"

SQL 语句对我来说看起来不错，所以参数。我错过了什么吗？

标签： pythonmysqlsqlalchemyflask-sqlalchemy

解决方案

如果你想:named_parameter为你的 SQL 语句使用表单，你需要使用 SQLAlchemy 的text方法，然后调用对象的execute方法Connection：

import sqlalchemy as sa

# ...

with engine.begin() as conn:
    conn.exec_driver_sql("CREATE TEMPORARY TABLE temp (id varchar(10) PRIMARY KEY)")
    sql = sa.text("INSERT INTO temp (id) VALUES (:id)")
    params = {'id': 'foo'}
    conn.execute(sql, params)

    print(conn.execute(sa.text("SELECT * FROM temp")).fetchall())  # [('foo',)]

推荐阅读