我有一个带有 SQLite 的 dev db（也可以运行单元测试）和带有 MySQL 的 prod db。
我需要使用 SQLAlchemy 编写一个查询，该查询在 WHERE 语句中使用子字符串。我试图使用func，但它在不修改特定数据库引擎的情况下翻译它。
我的查询是：
MyTable.field == func.substring_index(OtherTable.other_field, ":", 1)
所以我基本上想用“：”分割一个值并取第一部分。
问题是它被substring_index 翻译为SQLite，这是不正确的。有没有办法在 WHERE 子句中使用子字符串？
【问题讨论】：
标签： python sqlalchemy
SQLAlchemy 支持custom SQL constructs and compilation extensions 和registering named functions。有了这些，您可以将substring_index() 注册为对 SQLite 进行特殊处理的函数：
from sqlalchemy.sql.functions import GenericFunction
from sqlalchemy.types import String
from sqlalchemy.ext.compiler import compiles
class substring_index(GenericFunction):
type = String
@compiles(substring_index, 'sqlite')
def compile_substring_index_sqlite(element, compiler, **kw):
s, delim, count = element.clauses
# This assumes that count is a `bindparam`, produced from passing
# literal integer to `func.substring_index()`.
assert count.value == 1, "INSTR(X, Y) only supports first occurrence"
s = compiler.process(s, **kw)
delim = compiler.process(delim, **kw)
return f"substr({s}, 1, instr({s}, {delim}) - 1)"
另一种选择是在 SQLite 中将register a Python function 设置为substring_index()：
from sqlalchemy import event
def sqlite_substring_index(s, delim, count):
parts = s.split(delim)
if count > 0:
parts = parts[:count]
else:
parts = parts[count:]
return delim.join(parts)
# In your SQLite branch, before anything else DB related is performed:
@event.listens_for(engine, 'connect')
def create_functions(dbapi_connection, connection_record):
dbapi_connection.create_function('substring_index', 3, sqlite_substring_index)
有了这个函数，你就可以像在 MySQL 中那样调用它了。
至于为什么这不是开箱即用的 SQLAlchemy 库的一部分，由于不同的 DBMS 支持截然不同的功能，这将是一场无休止的战斗。例如，一些三角函数的名称不同，SQLite does not provide them at all 开箱即用。在单个代码库中支持不同的 SQL DBMS 并非易事，而且往往不值得。
【讨论】：
您可以检查数据库方言名称，并在此基础上创建子字符串。例如：
def substring(column, delimeter):
if session.bind.dialect.name == 'sqlite':
return func.substr(column, 1, func.instr(column, delimeter) - 1)
elif session.bind.dialect.name == 'mysql':
return func.substring_index(column, delimeter, 1)
然后将您的过滤条件替换为：
MyTable.field == substring(OtherTable.other_field, ":")
【讨论】：