如何在sqlalchemy的sqlite中使用regexp函数？

我想在“sqlalchemy”中使用 regexp查询以及在“ 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())

我发现应该在每个线程上注册regexp函数：

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

但是我不能对我的脚本采用链接的解决方案而已弃用.

更新

我想查询一下：

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

但在sqlalchemy.

解决方法

我有答案..
丢失一行的完整工作脚本是这样的：

import os
import re
import sqlite3

DB_PATH = __name__ + '.db'

try:
    os.remove(DB_PATH)
except:
    pass


def re_fn(expr,)
    ]
)
SEARCH_TERM = '3rd part'
cursor.execute(
    #'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',re_fn)

db = orm.scoped_session(orm.sessionmaker(autocommit=False,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 ..”

也许是因为这种变化：

When a file-based database is specified,the dialect will use NullPool
as the source of connections. This pool closes and discards
connections which are returned to the pool immediately. sqlite
file-based connections have extremely low overhead,so pooling is not
necessary. The scheme also prevents a connection from being used again
in a different thread and works best with sqlite’s coarse-grained file
locking.
Changed in version 0.7: Default selection of NullPool for sqlite
file-based databases. PrevIoUs versions select SingletonThreadPool by
default for all sqlite databases.

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

解决方案是：
在’begin’事件中添加regexp fn,如下所示：

...

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

...

如何在sqlalchemy的sqlite中使用regexp函数？

解决方法

猜你在找的Sqlite相关文章