检查 <sqlalchemy.engine.result.ResultProxy> 对象是否为空

Question

我正在通过一个使用后端的pymssql库从SQL Server数据库下载一些数据。 cursor.execute("""<QUERY BODY>""")的结果是sqlalchemy.engine.result.ResultProxy对象。如何检查查询结果是否为空，以便没有任何行？

cur = ff.sql.create_engine(server=dw.address, db=dw.BI_DW,
                               login=":".join([os.environ["SQL_USER"],
                                               os.environ["SQL_PASSWD"]]))
for n in range(100):
    result = cur.execute("""QUERY BODY;""")
    if result:
        break

很遗憾，即使SQL查询未返回任何行，result也永远不会是None。

那么最好的检查方法是什么？

- Gianluca

---

你能提供查询文本吗？如果你只需要检查获取的行数是否为零，那么可能有更有效的方法来实现这一点。 - arrakis_sun

查询体并不重要。我遇到的问题是SQL表没有使用昨天的数据进行分区，因此没有返回任何行。我需要在我的脚本中实现一些逻辑，以了解获取的行数是否为零。 - Gianluca

为什么你不能在查询中直接使用 COUNT(*)？ - arrakis_sun

for循环是一种（诚然愚蠢的）方法，用于查询数据库直到获得一些结果。我将对此进行重构。 - Gianluca

我不喜欢计算行数的解决方案，因为这意味着我需要运行两个查询。一个是用来计算行数并在必要时停止程序，另一个是获取结果，然后逐行处理。 - Gianluca

关于 ResultProxy 的问题在于，你无法知道返回的行数，直到你通过迭代一个个获取它们，或者通过 .fetchall() 方法一次性获取所有行。 - arrakis_sun

Answer 1

ResultProxy对象当前并不包含任何行。因此，它没有关于总行数甚至是否有任何行的信息。 ResultProxy只是指向数据库的“指针”。 只有在通过ResultProxy显式获取行时，您才会得到行。 您可以通过迭代此对象、使用.first()方法或.fetchall()方法来达到这个目的。

总之：直到您实际提取所有行且ResultProxy对象已耗尽，您才能确定提取行的数量。

方法#1

您可以一次获取所有行并对其进行计数，然后执行所需操作：

rows = result.fetchall()
if len(rows):
    # do something with rows

这种方法的缺点是我们一次性将所有行加载到内存中（rows 是包含所有获取的行的 Python list）。如果获取的行数非常大和/或者你只需要独立逐个迭代每一行（通常是这种情况），那么这可能不是一个理想的选择。

方法 #2

如果一次性将所有获取的行加载到内存中不可接受，那么我们可以这样做：

rows_amount = 0
for row in result:
    rows_amount += 1
    # do something with row
if not rows_amount:
    print('There were zero rows')
else:
    print('{} rows were fetched and processed'.format(rows_amount))

- arrakis_sun

---

我甚至没有考虑过 first()。然而，我需要在此检查之后处理这些行，所以这个解决方案对我来说不可行。 - Gianluca

sqlalchemy.exc.ResourceClosedError: 此结果对象不返回行。它已被自动关闭。我按照您的评论使用了fetchone()和fetchall，但仍然出现了以上错误。 - Mahmoud Magdy

Answer 2

SQLAlchemy < 1.2: 你总是可以将 ResultProxy 转换为迭代器：

res = engine.execute(...)
rp_iter = iter(res)
row_count = 0
try:
    row = next(rp_iter)
    row_count += 1
except StopIteration:
    # end of data
    if not row_count:
      # no rows returned, StopIteration was raised on first attempt

在 SQLAlchemy >= 1.2 中，ResultProxy 实现了 .next() 和 .__next__()，因此您无需创建迭代器：

res = engine.execute()
row_count = 0
try:
    row = next(res)
    row_count += 1
except StopIteration:
    ...

- cowbert