How to use regexp function in sqlite with sqlalchemy?

Question

I would like to use the regexp query in "sqlalchemy" as well as in "python sqlite", the code below ..

The incomplete sandbox script is as follows:

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()) 

I found that the regexp function must be registered in each thread:

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

but I cannot decide to reference my script +, which it is deprecated.

Refresh

I would like to request the following:

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

but in sqlalchemy.

Answer 1

I have an answer. The full script with one line missing is:

 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()) 

Above works in sqlalchemy = 0.6.3

In sqlalchemy = 0.7.8, I got an error:

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

possibly due to this change:

When a file database is specified, the dialect will use NullPool as the source of the links. This pool is closed and discarded which immediately returns to the pool. SQLite file connections have extremely low overhead, so merging is not necessary . This scheme also prevents reuse of the connection in another thread and works best with the coarse-grained SQLites lock file. Changed in version 0.7: default selection for NullPool for file-based SQLite. Previous versions chose SingletonThreadPool by default for all SQLite databases.

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

and the solution for this was: add regexp fn to the "begin" event as follows:

 ... 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)) ...