how to exec sql directly

[MrYZhou]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

@router.get("/dynamicConnect")
async def dynamicConnect():
    sql = """
    SELECT TB.TABLE_NAME as dbName,TB.TABLE_COMMENT as tableComment, COL.COLUMN_NAME as columnName,COL.COLUMN_COMMENT as columnComment,COL.DATA_TYPE   as dataType
FROM INFORMATION_SCHEMA.TABLES TB,INFORMATION_SCHEMA.COLUMNS COL
Where TB.TABLE_SCHEMA ='study' AND TB.TABLE_NAME = COL.TABLE_NAME
    """
    DB_URL = "mysql+pymysql://root:123456@localhost:3307/study?charset=utf8mb4"
    engine = create_engine(DB_URL)
    with Session(engine) as session:
        list = session.execute(sql).fetchall()
        return list

Description

The previous version of sqlmodel was executable, but the new version suggested that I use exec. However, I found that exec cannot implement my original function. How should I modify it

Operating System

Windows

Operating System Details

No response

SQLModel Version

0.0.14

Python Version

Python 3.11.2

Additional Context

No response

[connorkeevill]

Try using the SQLAlchemy text() function (see here: https://www.tutorialspoint.com/sqlalchemy/sqlalchemy_core_using_textual_sql.htm) alongside the old style Session.execute() function.

So your example would look something like:

from sqlalchemy import text

@router.get("/dynamicConnect")
async def dynamicConnect():
    sql = text("""
    SELECT TB.TABLE_NAME as dbName,TB.TABLE_COMMENT as tableComment, COL.COLUMN_NAME as columnName,COL.COLUMN_COMMENT as columnComment,COL.DATA_TYPE   as dataType
FROM INFORMATION_SCHEMA.TABLES TB,INFORMATION_SCHEMA.COLUMNS COL
Where TB.TABLE_SCHEMA ='study' AND TB.TABLE_NAME = COL.TABLE_NAME
    """)
    DB_URL = "mysql+pymysql://root:123456@localhost:3307/study?charset=utf8mb4"
    engine = create_engine(DB_URL)
    with Session(engine) as session:
        list = session.execute(sql).fetchall()
        return list

[MrYZhou]

@router.get("/dynamicConnect")
async def dynamicConnect():
sql = """SELECT TB.TABLE_NAME as dbName,TB.TABLE_COMMENT as tableComment, COL.COLUMN_NAME as columnName,COL.COLUMN_COMMENT as columnComment,COL.DATA_TYPE as dataType
FROM INFORMATION_SCHEMA.TABLES TB,INFORMATION_SCHEMA.COLUMNS COL
Where TB.TABLE_SCHEMA ='study' AND TB.TABLE_NAME = COL.TABLE_NAME"""
DB_URL = "mysql+pymysql://root:root@localhost:3306/study?charset=utf8mb4"
engine = create_engine(DB_URL)
with Session(engine) as session:
# list:List[Table] = session.execute(sql).fetchall()
list:List[Table] = session.exec(select(text(sql))).all()
return list

but,Still no, there is also a problem with json conversion, which must have type annotations.

I want to emphasize that session.execute has been deprecated, otherwise I wouldn't have had such a problem with the old version, and the link you gave is still the deprecated solution.

[darrenclark]

+1 to having an officially supported way of running SQL directly.

I've had success with something like this to at least hide the deprecation warning:

sql = "SELECT ......."
await super(AsyncSession, async_session).execute(text(sql))

Haven't tested it, but I assume similar would work for non-async:

sql = "SELECT ......."
super(Session, session).execute(text(sql))

[YuriiMotov]

The absence of proper overload for exec to work with text (and also insert, update, delete) statement is known.
There is issue to track that: #909