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

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:

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()

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

Thanks in advance!

CodePudding user response：

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()