python - 将 MySQL 查询转换为 SQLalchemy 查询

Question

我正在尝试从使用简单 BASH 脚本运行的服务器移植所有原始 SQL 脚本，以便我可以使用 python 和 sqlalchemy 处理数据，但下面的这个脚本拒绝移植。我还在下面列出了我复制它的最接近的尝试。

如果有人对我的错误有任何想法，或者您可以复制它，请随意。

另请注意，某些表已被替换为填充名称。

-----MYSQL-----

SELECT *
FROM
(SELECT CONCAT('{URL}', TBL1.ColumID) AS 'Link',
    TBL1.ColumID AS CaseID,
    TBL1.ColumType AS 'Abuse Type',
    TBL1.Status AS 'Status',
    TBL1.Role AS 'Queue', TBL1.CreationDate, TBL1.LastUpdated, @LAST_TOUCHED:=
    (SELECT from_unixtime((max(unix_timestamp(LAST_UPDATED))))
        FROM TBL2
        WHERE TBL1.ColumID = CaseID
        AND (NOTE_TYPE = 'communication'
            OR NOTE_SUBTYPE IS NULL)) AS last_touched, ((unix_timestamp(now()) - unix_timestamp(LAST_UPDATED))/3600/24) AS Age,
    CASE
    WHEN isnull(@LAST_TOUCHED) THEN 'N'
    ELSE 'Y'
    END AS Touched
    FROM TBL1
    WHERE TBL1.ROLE REGEXP '{ROLE Filler}'
    AND STATUS REGEXP 'RESOLVED|DELETED' =0) AS tbl
WHERE tbl.Age > 2
order by tbl.Age desc

------失败的Python代码-----

query = select([
        TBL1.c.caseid.label('CaseID'),
        TBL1.c.casetype.label('Type'),
        TBL1.c.STATUS.label('Status'),
        TBL1.c.ROLE.label('Queue'),
        TBL1.c.CREATION_DATE,
        TBL1.c.LAST_UPDATED,
        (func.from_unixtime(
            func.max(
                func.unix_timestamp(
                    TBL2.c.LAST_UPDATED))).filter(
        TBL2.c.caseid == TBL1.c.caseid).filter(
        (TBL2.c.NOTE_TYPE == '{notetype}') | 
        (TBL2.c.NOTE_SUBTYPE == None)).label("LAST_TOUCHED")
        )]).where(
        (TBL1.c.ROLE == '{rolename}') &
        (TBL1.c.STATUS != 'RESOLVED') |
        (TBL1.c.STATUS != 'DELETED'))

Answer 1

首先，仔细检查您是否确实必须这样做。您这样做是为了将查询包含在网络服务器中，还是为了提取数据进行分析？如果是后者，只需使用pandas:

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('{connection string}')
df = pd.read_sql_query("{insert your complicated query here}", engine)