6

「sqlalchemy」で正規表現クエリを使用したいのですが、以下のコードの「pythonsqlite」で実行します。

未完成のサンドボックススクリプトは次のとおりです。

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では。

4

2 に答える 2

13

私は答えを持っています..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）そのような関数はありません：regexp.."

おそらくこの変更のため：

ファイルベースのデータベースが指定されている場合、方言は接続のソースとしてNullPoolを使用します。このプールは閉じて、すぐにプールに戻される接続を破棄します。SQLiteファイルベースの接続はオーバーヘッドが非常に低いため、プーリングは必要ありません。このスキームはまた、接続が別のスレッドで再び使用されるのを防ぎ、SQLiteの粗粒度のファイルロックで最適に機能します。バージョン0.7で変更：SQLiteファイルベースのデータベースのNullPoolのデフォルトの選択。以前のバージョンでは、すべてのSQLiteデータベースに対してデフォルトでSingletonThreadPoolが選択されていました。

差出人：http ：//docs.sqlalchemy.org/en/rel_0_7/dialects/sqlite.html？highlight = isolation_level＃threading-pooling-behavior

そのための解決策は次のとおりです。次のように「begin」イベントに正規表現fnを追加します。

...

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

...
于 2012-09-18T10:18:15.117 に答える
0

ほぼ10年後...

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

于 2022-01-20T22:43:38.443 に答える