如何在SQLAlchemy中使用SQLite的正则表达式函数（regexp函数）？

Question

我希望您能在"sqlalchemy"中像在"python sqlite"中一样使用正则表达式查询，以下是代码：

未完成的沙盒脚本如下：

import os
import re
import sqlite3

#
# python sqlite
#

DB_PATH = __name__ + '.db'

try:
    os.remove(DB_PATH)
except:
    pass


def re_fn(expr, item):
    reg = re.compile(expr, re.I)
    return reg.search(item) is not None

conn = sqlite3.connect(':memory:')
conn = sqlite3.connect(DB_PATH)
conn.create_function("REGEXP", 2, re_fn)
cursor = conn.cursor()

cursor.execute(
    'CREATE TABLE t1 (id INTEGER PRIMARY KEY, c1 TEXT)'
)
cursor.executemany(
    #'INSERT INTO t1 (c1) VALUES (?)', [('aaa"test"',),('blah',)]
    'INSERT INTO t1 (c1) VALUES (?)', [
        ('dupa / 1st Part',), ('cycki / 2nd Part',), ('fiut / 3rd Part',)
    ]
)
cursor.execute(
    #'SELECT c1 FROM t1 WHERE c1 REGEXP ?',['2|3\w+part']
    'SELECT c1 FROM t1 WHERE c1 REGEXP ?',['\d\w+ part']
)
conn.commit()
data=cursor.fetchall()
print(data)



#
# sqlalchemy
#

import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy.ext.declarative import declarative_base

DSN = 'sqlite:///' + DB_PATH
engine = sa.create_engine(DSN, convert_unicode=True)
db = orm.scoped_session(orm.sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))

Base = declarative_base(bind=engine)
meta = Base.metadata

class T1(Base):
    __table__ = sa.Table('t1', meta, autoload=True)

print(db.query(T1).all())

我发现正则表达式函数应该在每个线程上注册：

http://permalink.gmane.org/gmane.comp.web.pylons.general/12742

但我无法采用链接中的解决方案到我的脚本中，而且它已经过时。

更新

我想查询这个：

cursor.execute(
    #'SELECT c1 FROM t1 WHERE c1 REGEXP ?',['2|3\w+part']
    'SELECT c1 FROM t1 WHERE c1 REGEXP ?',['\d\w+ part']
)

但在SQLAlchemy中。

- xliiv

---

我不明白你想要做什么。你有哪些错误？ - njzk2

Answer 1

我知道答案了... 缺失一行的完整工作脚本如下：

import os
import re
import sqlite3

DB_PATH = __name__ + '.db'

try:
    os.remove(DB_PATH)
except:
    pass


def re_fn(expr, item):
    reg = re.compile(expr, re.I)
    return reg.search(item) is not None

conn = sqlite3.connect(':memory:')
conn = sqlite3.connect(DB_PATH)
conn.create_function("REGEXP", 2, re_fn)
cursor = conn.cursor()

cursor.execute(
    'CREATE TABLE t1 (id INTEGER PRIMARY KEY, c1 TEXT)'
)
cursor.executemany(
    #'INSERT INTO t1 (c1) VALUES (?)', [('aaa"test"',),('blah',)]
    'INSERT INTO t1 (c1) VALUES (?)', [
        ('dupa / 1st Part',), ('cycki / 2nd Part',), ('fiut / 3rd Part',)
    ]
)
SEARCH_TERM = '3rd part'
cursor.execute(
    #'SELECT c1 FROM t1 WHERE c1 REGEXP ?',['2|3\w+part']
    'SELECT c1 FROM t1 WHERE c1 REGEXP ?',[SEARCH_TERM]
)
conn.commit()
data=cursor.fetchall()
print(data)



#
# sqlalchemy
#

import sqlalchemy as sa
import sqlalchemy.orm as orm
from sqlalchemy.ext.declarative import declarative_base

DSN = 'sqlite:///' + DB_PATH

engine = sa.create_engine(DSN, convert_unicode=True)

conn = engine.connect()
conn.connection.create_function('regexp', 2, re_fn)

db = orm.scoped_session(orm.sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))

Base = declarative_base(bind=engine)
meta = Base.metadata

class T1(Base):
    __table__ = sa.Table('t1', meta, autoload=True)

print(db.query(T1.c1).filter(T1.c1.op('regexp')(SEARCH_TERM)).all())

在 sqlalchemy=0.6.3 中以上代码可正常运行。

但在 sqlalchemy=0.7.8 中我遇到了错误:

"sqlalchemy.exc.OperationalError: (OperationalError) no such function: regexp .."

可能是因为这个改变:

当指定基于文件的数据库时，方言将使用 NullPool 作为连接源。 该池会立即关闭和丢弃返回到池中的连接。SQLite 基于文件的连接具有极低的开销，因此不需要池化。该方案还防止连接在不同线程中再次使用，并且与 SQLite 的粗粒度文件锁定最佳配合使用。 在版本 0.7 中更改：针对 SQLite 基于文件的数据库，默认选择 NullPool。之前的版本默认选择 SingletonThreadPool 用于所有 SQLite 数据库。

来源: http://docs.sqlalchemy.org/en/rel_0_7/dialects/sqlite.html?highlight=isolation_level#threading-pooling-behavior

解决方法是：在“begin”事件中添加正则表达式函数，如下所示：

...

conn = engine.connect()
@sa.event.listens_for(engine, "begin")
def do_begin(conn):
    conn.connection.create_function('regexp', 2, re_fn)

db = orm.scoped_session(orm.sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))

...

- xliiv

---

我不明白“re_fn”从哪里获取正则表达式模式？你只将搜索项作为参数放入，但该函数要求两个参数。谢谢！ - TmSmth

我猜item参数是由SQL引擎注入的行。因此，程序员指定正则表达式（如“\d+”），而SQL引擎注入数据库行。有了这两个参数，您可以在re_fn中决定是否返回特定行。 - xliiv

哦，好的，我明白了，“3rd part”是“expr”，不是“item”！谢谢。 - TmSmth

Answer 2

近10年过去了...

在SQLAlchemy 1.4中，它更容易工作：

stmt = select(table.c.some_column).where(
    table.c.some_column.regexp_match('^(b|c)')
)

来源：https://docs.sqlalchemy.org/en/14/core/sqlelement.html#sqlalchemy.sql.expression.ColumnElement.regexp_match

- dlazesz