如何使用 sqlite3 连接游标执行 sqlalchecmy TextClause 语句?

How to execute a sqlalchecmy TextClause statement with a sqlite3 connection cursor?

我有一个 python flask 应用程序,它主要使用 sqlalchemy 来执行它的所有 mySQL 查询,我需要使用本地数据库和行为为其编写测试。

经过简单的研究,我为这个任务选择的数据库是本地 sqlite3 db,主要是因为我读到它与 mySQL 和 sqlalchemy 非常兼容,也因为它很容易设置和拆除。 我已成功建立连接并成功创建了测试所需的所有 tables。

我在尝试执行某些查询时遇到问题,其中查询语句被构建为 sqlalchemy TextClause 对象,而我的 sqlite3 连接游标在尝试执行语句时引发以下异常:

TypeError: argument 1 must be str, not TextClause

如何将此 TextClause 对象动态转换为字符串并执行? 我不想为了测试而对代码进行重大更改。

代码示例: 员工 table:

id name
1 Jeff Bezos
2 Bill Gates
from sqlalchemy import text
import sqlite3

def select_employee_by_id(id: int):
    employees_table = 'employees'
    db = sqlite3.connect(":memory:")
    cursor = db.cursor()
    with db as session:
        statement = text("""
                            SELECT  *
                            FROM {employees_table}
                            WHERE
                                id = :id
                        """.format(employees_table=employees_table)
                         ).bindparams(id=id)
        data = cursor.execute(statement)
        return data.fetchone()

应该 return 一行包含 {'id': 1, 'name': 'Jeff Bezos'} for select_employee_by_id(1)

提前致谢!

如果您想测试 TextClause 查询,那么您应该使用 SQLAlchemy 执行它,而不是使用 DBAPI (SQLite) 游标:

from sqlalchemy import create_engine, text

def select_employee_by_id(id: int):
    employees_table = 'employees'
    engine = create_engine("sqlite://")
    with engine.begin() as conn:
        statement = text("""
                            SELECT  *
                            FROM {employees_table}
                            WHERE
                                id = :id
                        """.format(employees_table=employees_table)
                         ).bindparams(id=id)
        data = conn.execute(statement)
        return data.one()