基于SQLAlchemy的Database线程类select语句未执行问题排查

导致SQL未执行的核心问题 #

1. 主线程被join()永久阻塞 #

你的Database类初始化方法中，启动子线程后立刻执行self.join()，这会强制主线程等待子线程完全终止后再继续。但子线程的run()方法里是无限循环（只有收到--close--指令才会退出），所以主线程会一直卡在__init__阶段，后续调用get_p、执行SQL的代码根本没机会运行。

修复： 移除__init__中的self.join()，让子线程在后台独立运行：

class Database(threading.Thread):
    def __init__(self, db, logger = None):
        super(Database, self).__init__()
        self.db=db
        self.reqs=Queue()
        self.log = logger
        self.start()
        # 删除此行：self.join()

2. 生成器未被触发，SQL逻辑未执行 #

select方法返回的是生成器对象，而print(get_p(dbstore))只会打印生成器的内存地址，不会触发生成器内部的队列请求、结果获取逻辑。生成器是惰性执行的，必须迭代它才会运行内部代码。

修复： 将生成器转为列表（或手动迭代）来触发执行：

def get_p(found:Database):
    data = found.select("SELECT count(*) FROM products")
    return list(data)  # 迭代生成器，触发SQL执行与结果返回

3. 数据库连接被提前关闭 #

getconn()中使用with Connector() as connector，当with块结束时，connector的__exit__方法会被调用，大概率会关闭连接或连接池，导致返回的conn已经失效，后续使用时可能无法正常执行SQL。

修复： 去掉with语句，直接创建连接器并返回连接：

def getconn():
    connector = Connector()
    conn = connector.connect(
        instance_connection_name,
        "pg8000",
        user = db_user,
        password = db_pass,
        db = db_name,
        ip_type = IPTypes.PUBLIC
    )
    return conn

修复后的关键代码示例 #

修改后的Database类初始化 #

import threading
import sqlite3
from queue import Queue
import sqlalchemy

class Database(threading.Thread):
    def __init__(self, db, logger = None):
        super(Database, self).__init__()
        self.db=db
        self.reqs=Queue()
        self.log = logger
        self.start()  # 仅启动子线程，不阻塞主线程
    
    # 其余方法保持不变...

修改后的调用代码 #

# initialize Connector object
connector = Connector()

def getconn():
    connector = Connector()
    conn = connector.connect(
        instance_connection_name,
        "pg8000",
        user = db_user,
        password = db_pass,
        db = db_name,
        ip_type = IPTypes.PUBLIC
    )
    return conn

def get_p(found:Database):
    data = found.select("SELECT count(*) FROM products")
    return list(data)  # 触发生成器执行

dbstore = Database(db=getconn)
try:
    print(get_p(dbstore))
except Exception as e:
    print(e)
# 程序结束时记得关闭数据库线程
# dbstore.close()

额外提示 #

• 程序退出前务必调用dbstore.close()，让子线程退出循环并关闭数据库连接，避免资源泄漏。
• 确认SQLAlchemy引擎的creator参数配置正确：self.db是getconn函数，符合要求（返回DBAPI连接的可调用对象）。

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