如何用SQLAlchemy流式查询非NULL的coordinates字段数据？

需求与原问题 #

需要筛选表中coordinates列不为NULL的所有行的主键，但原使用select()查询时因数据量过大导致内存耗尽：

s = tweets.select().where(tweets.c.coordinates != None)

result = engine.execute(s)

for row in result:
    print(row)

尝试的分页方案 #

找到适用于session.query()的流式分页函数：

def page_query(q):
    r = True
    offset = 0
    while r:
        r = False
        for elem in q.limit(1000).offset(offset):
            r = True
            yield elem
        offset += 1000

筛选条件遇到的问题 #

改用session.query()后出现两个问题：

• 使用q= session.query(Tweet).filter(Tweet.coordinates.is_not(None))会返回所有行（包括coordinates为null的行）
• 使用q= session.query(Tweet).filter(Tweet.coordinates.is_not('null'))触发语法错误：

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.SyntaxError) syntax error at or near "'null'"
LINE 3: WHERE milan_tweets.coordinates IS NOT 'null' 
                                              ^

补充：表中数据以Python None存入，在DBeaver中显示为"null"

正确解决方案 #

1. 确认数据实际存储类型 #

先排查coordinates列的实际存储值，避免混淆SQL NULL与字符串"null"：

# 替换成你数据库中显示为null的记录ID
test_row = session.query(Tweet.coordinates).filter(Tweet.id == 123).first()
print(type(test_row[0]), test_row[0])

• 如果输出为<class 'NoneType'> None：说明存储的是SQL NULL，用下面的筛选条件
• 如果输出为<class 'str'> null：说明实际存入的是字符串"null"，用字符串匹配的筛选条件

2. 对应筛选条件 #

存储为SQL NULL的情况

使用SQLAlchemy标准的非NULL筛选写法：

# 两种写法等价，任选其一
q = session.query(Tweet).filter(Tweet.coordinates.isnot(None))
# 或
q = session.query(Tweet).filter(Tweet.coordinates != None)

如果仍返回所有行，检查模型中coordinates字段是否设置了nullable=True，确保字段允许存储NULL值。

存储为字符串"null"的情况

直接用字符串不等于的条件：

q = session.query(Tweet).filter(Tweet.coordinates != 'null')

3. 更高效的内存优化方案 #

除了自定义分页，还可以直接给查询开启流式返回，避免一次性加载所有数据：

# 原select查询的流式写法
s = tweets.select().where(tweets.c.coordinates != None)
result = engine.execute(s.execution_options(stream_results=True))
for row in result:
    print(row)

# session.query的流式写法
q = session.query(Tweet).filter(Tweet.coordinates.isnot(None))
result = q.execution_options(stream_results=True).yield_per(1000)
for elem in result:
    print(elem)

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