SQLAlchemy操作BigQuery时能否给查询对象追加原生SQL过滤片段？

可以直接追加原生SQL片段作为过滤条件 #

你可以使用SQLAlchemy内置的text()方法包装原生SQL片段，再传入filter()方法即可追加到现有查询对象中，SQLAlchemy会自动将新条件用AND逻辑拼接到原有WHERE子句中，无需手动添加开头的AND关键字。

实现示例：

1. 先导入依赖的text方法

from sqlalchemy import text

2. 追加自定义过滤条件

# 原有查询逻辑
query = TestModel.query.filter(TestModel.timestamp.__gt__(1010101010))
# 追加EXISTS原生过滤逻辑
query = query.filter(text("EXISTS (SELECT * FROM UNNEST(column_name) WHERE column_key LIKE '%tit%')"))

动态参数安全实现建议：

如果你的匹配关键词是动态可变的，不要直接拼接字符串避免SQL注入风险，可以使用占位符+参数绑定的方式实现：

from sqlalchemy import text

search_key = "tit"
query = query.filter(
    text("EXISTS (SELECT * FROM UNNEST(column_name) WHERE column_key LIKE :pattern)")
).params(pattern=f"%{search_key}%")

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