Python使用SQLAlchemy时MySQL查询空结果致程序挂起问题求助

先修复代码中的基础错误 #

你的代码存在几个明显的语法/逻辑错误，先修正这些才能进一步解决挂起问题：

1. customers_df.itterows()拼写错误，应为customers_df.iterrows()
2. row('id')调用错误，Pandas的行数据是Series，应该用row['id']
3. SQL排序逻辑错误：你要找最接近结束日期的订单，应该按orderDate DESC排序，而不是customerId DESC，否则排序毫无意义

核心问题：空结果导致Pandas挂起的解决思路 #

方案1：用SQLAlchemy Result对象直接处理查询结果

绕过pd.readsql，直接用SQLAlchemy执行查询并处理空结果，这样能完全控制流程，避免Pandas的等待问题：

import pandas as pd
from sqlalchemy import text

customer_price_at_date = pd.DataFrame(columns=['customerId', 'purchasePrice', 'orderPrice', 'orderDate'])

startDate = '2020-02-11'
endDate = '2020-02-28'

loop_len = len(customers_df)
# 复用连接，不要每次循环创建，SQLAlchemy默认带连接池
with engine.connect() as db_connection:
    for index, row in customers_df.iterrows():
        customer_id = row['id']
        try:
            # 使用参数化查询，避免SQL注入，同时更安全
            query = text("""
                SELECT customerId, purchasePrice, orderPrice, orderDate 
                FROM orders 
                WHERE customerId = :cid AND orderDate BETWEEN :start AND :end 
                ORDER BY orderDate DESC LIMIT 1
            """)
            result = db_connection.execute(query, {"cid": customer_id, "start": startDate, "end": endDate})
            row_data = result.one_or_none()  # 空结果返回None，不会挂起
            
            if row_data:
                # 将结果转为DataFrame行并追加
                customer_price_at_date = pd.concat([
                    customer_price_at_date,
                    pd.DataFrame([row_data._asdict()])
                ], ignore_index=True)
            
            loop_len -= 1
            print(f'Processing Customer Id {customer_id} | Remaining: {loop_len}')
            
        except Exception as e:
            print(f'Error processing Customer Id {customer_id}: {str(e)} | Remaining: {loop_len}')
            continue

方案2：设置查询超时，强制终止无响应的查询

在创建SQLAlchemy引擎时，添加MySQL的超时参数，确保即使出现异常情况（包括空结果导致的无响应）也能在指定时间内抛出异常，被try-except捕获：

# 创建引擎时添加超时配置
from sqlalchemy import create_engine

engine = create_engine(
    'mysql+pymysql://user:password@host/dbname',
    connect_args={
        "connect_timeout": 30,  # 连接超时30秒
        "read_timeout": 300     # 读取超时5分钟，对应你说的5分钟
    }
)

# 后续代码可以用修正后的pd.readsql，但依然推荐参数化查询
customer_price_at_date = pd.DataFrame()
startDate = '2020-02-11'
endDate = '2020-02-28'
loop_len = len(customers_df)

with engine.connect() as db_connection:
    for index, row in customers_df.iterrows():
        customer_id = row['id']
        try:
            query = text("""
                SELECT customerId, purchasePrice, orderPrice, orderDate 
                FROM orders 
                WHERE customerId = :cid AND orderDate BETWEEN :start AND :end 
                ORDER BY orderDate DESC LIMIT 1
            """)
            # 用readsql的params参数传参
            df = pd.readsql(query, con=db_connection, params={"cid": customer_id, "start": startDate, "end": endDate})
            if not df.empty:
                customer_price_at_date = pd.concat([customer_price_at_date, df], ignore_index=True)
            
            loop_len -= 1
            print(f'Processing Customer Id {customer_id} | Remaining: {loop_len}')
            
        except Exception as e:
            print(f'Error processing Customer Id {customer_id}: {str(e)} | Remaining: {loop_len}')
            continue

额外性能优化建议 #

1. 不要每次循环创建/关闭连接：SQLAlchemy默认使用连接池，复用连接能大幅提升2万次循环的执行效率
2. 使用参数化查询：避免字符串拼接带来的SQL注入风险，同时数据库能缓存查询计划，提升重复查询的速度
3. 批量查询替代循环：如果允许，直接一次性查询所有目标客户的最新订单，比如用窗口函数ROW_NUMBER()按客户分组排序，这样只需一次查询，效率远高于2万次循环：

SELECT customerId, purchasePrice, orderPrice, orderDate
FROM (
    SELECT 
        customerId, purchasePrice, orderPrice, orderDate,
        ROW_NUMBER() OVER (PARTITION BY customerId ORDER BY orderDate DESC) AS rn
    FROM orders
    WHERE customerId IN (:customer_ids) AND orderDate BETWEEN :start AND :end
) t
WHERE rn = 1

然后在Python中一次性传入所有客户ID：

customer_ids = customers_df['id'].tolist()
query = text("""
    SELECT customerId, purchasePrice, orderPrice, orderDate
    FROM (
        SELECT 
            customerId, purchasePrice, orderPrice, orderDate,
            ROW_NUMBER() OVER (PARTITION BY customerId ORDER BY orderDate DESC) AS rn
        FROM orders
        WHERE customerId IN (:customer_ids) AND orderDate BETWEEN :start AND :end
    ) t
    WHERE rn = 1
""")
customer_price_at_date = pd.readsql(query, con=engine, params={"customer_ids": customer_ids, "start": startDate, "end": endDate})

这种方式能把2万次查询压缩为1次，性能提升几个数量级，强烈推荐。

内容的提问来源于stack exchange，提问作者DuncanG