如何在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|3w+part'] 'SELECT c1 FROM t1 WHERE c1 REGEXP ?',['dw+ 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 ?',['dw+ 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我得到错误:
也许是因为这种变化:
来自:http://docs.sqlalchemy.org/en/rel_0_7/dialects/sqlite.html?highlight=isolation_level#threading-pooling-behavior 解决方案是: ... conn = engine.connect() @sa.event.listens_for(engine,"begin") def do_begin(conn): conn.connection.create_function('regexp',bind=engine)) ... (编辑:李大同) 【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容! |