How to use regexp function in sqlite with sqlalchemy?

Question

I'd like to use regexp query in "sqlalchemy" as well as is done in "python sqlite", code below..

Unfinished sandbox script is this:

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've found that regexp function should be registered on each thread:

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

but I'm not able to adopt link's solution to my script + it's deprecated.

Update

I'd like to query this:

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've got the answer.. Complete working script with missing one line is this:

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 error:

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

maybe because of this change:

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.

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

and solution for that was: to add regexp fn in 'begin' event like this:

...

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

...