如何使用SQLITE3连接光标执行SQLALCHECMY TEXTCLAUSE语句?

发布于 2025-02-01 09:15:54 字数 1316 浏览 2 评论 0原文

我有一个Python烧瓶应用程序,该应用主要使用SQLalchemy执行所有MySQL查询,我需要使用本地数据库为其编写测试并表现。

经过简短的研究,我为此任务选择的数据库是本地SQLite3 dB,主要是因为我已经读到它与MySQL和SQLalchemy兼容,也是因为它易于设置和拆除。 我已经成功建立了与它的连接,并设法创建了我对测试所需的所有表。

在尝试执行一些查询时,我遇到了一个问题,在其中构建查询语句作为sqlalchemy textclause对象,而我的sqlite3连接光标在尝试执行语句时提出了以下例外:

TypeError: argument 1 must be str, not TextClause

我如何可以动态转换此textclause对象转换为字符串并执行它? 我不想只是为了进行测试而对代码进行彻底的更改。

一个代码示例: 员工表:

ID名称
1Jeff Bezos
2Bill 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()

应返回包含{'ID':1,'名称'的行:'Jeff Bezos'},select_employeee_by_id(1)

谢谢!

I have a python flask app which primarily uses sqlalchemy to execute all of it's mySQL queries and I need to write tests for it using a local database and behave.

After a brief research, the database I've chosen for this task is a local sqlite3 db, mainly because I've read that its pretty much compatible with mySQL and sqlalchemy, and also because it's easy to set up and tear-down.
I've established a connection to it successfully and managed to create all the tables I need for the tests.

I've encountered a problem when trying to execute some queries, where the query statement is being built as a sqlalchemy TextClause object and my sqlite3 connection cursor raises the following exception when trying to execute the statement:

TypeError: argument 1 must be str, not TextClause

How can I convert this TextClause object dynamically to a string and execute it?
I don't want to make drastic changes to the code just for testing.

A code example:
employees table:

idname
1Jeff Bezos
2Bill 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()

Should return a row containing {'id': 1, 'name': 'Jeff Bezos'} for select_employee_by_id(1)

Thanks in advance!

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。

评论(1

看透却不说透 2025-02-08 09:15:54

如果要测试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()

If you want to test your TextClause query then you should execute it by using SQLAlchemy, not by using a DBAPI (SQLite) cursor:

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()
~没有更多了~
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文