在Python中的Azure SQL上使用CONTAINS和SQLAlchemy

Question

我正在尝试使用

CONTAINS

或

FREETEXT

功能进行全文搜索。

我使用 Azure SQL 和 SQLAlchemy，并使用 mssql 和 aioodbc 驱动程序。

我有一个包含 3 个文本列的表格，我想对其进行搜索。

我已经使用这些查询创建了目录、索引并启用了 CHANGE_TRACKING：

CREATE FULLTEXT CATALOG ftCatalog AS DEFAULT;

CREATE FULLTEXT INDEX ON myTable(
    col1,
    col2,
    col3
)
KEY INDEX PK__blabla_ans__3213E83F6A7DF9F1

ALTER FULLTEXT INDEX ON myTable SET CHANGE_TRACKING AUTO;

我可以像这样运行查询，它有效：

SELECT * FROM dbo.myTable
WHERE CONTAINS(col1, '"foo-bar"') OR CONTAINS(col2, '"foo-bar"') OR CONTAINS(col3, '"foo-bar"')

现在我尝试在 SQLAlchemy 中执行此操作，但它不想工作，这就是我尝试过的

1. .包含

   select(myTable).where(
     myTable.col1.contains('foo-bar')
   )
   

   导致 LIKE 运算符有效，但与 CONTAINS 不同。它也更慢......在 DBEaver 中运行自定义查询大约需要 0.03 秒，而此查询需要超过 1 秒。

2. 在

   select(myTable).where(
     'foo-bar' in myTable.col1
   )
   

   似乎想要使用 CONTAINS 但失败了：

   NotImplementedError：此表达式不支持运算符“包含”。

   这是驱动程序的问题吗？

3. 比赛

   select(myTable).where(
     myTable.col1.match('foo-bar')
   )
   

   也失败了：

   sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) ('42000', '[42000] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]参数类型“varchar(max)”对于参数 2 无效(4110) (SQLExecDirectW); [Microsoft][SQL Server 的 ODBC 驱动程序 18][SQL Server] 无法准备语句 (8180)')

我尝试将列转换为 nvarchar 但无法使其工作，不确定是否应该......

这在 SQLAlchemy 中是否可行，还是我应该尝试使用自定义原始查询？哪种违背了 ORM 的目的。

4. 这似乎有效：

func.freetext(myTable.col1, text(f"'{filters.search_term}'"))

但不是这个：

func.contains(myTable.col1, text(f"'{filters.search_term}'"))

Azure SQL 也支持这种语法，这对于 sqlalchemy 来说似乎是不可能的

WHERE CONTAINS((myTable.col1, myTable.col2, myTable.col3), 'foo-bar')

相反，我做 OR ，这看起来比较慢

or_(
  func.freetext(myTable.col1, text(f"'{filters.search_term}'"))
  func.freetext(myTable.col2, text(f"'{filters.search_term}'"))
)

Answer 1

啊是的，我明白了。

解决方案是列类型 NVARCHAR 并且搜索词需要转义，因此它也支持空格（包含多个单词的字符串）

使用如下参数防止 SQL 注入：

db_filters = list()
db_filters.append(
  func.contains(myTable.col1, text(f":search_term"))
)

并使用传入的参数运行查询，如下所示：

queryset = await session.execute(
  select(myTable)
  .where(*db_filters)
  .order_by(myTable.id.desc())
  , {"search_term": f"\"{search_term}\""}
)