[enh] utils: implement a context manager for a SQLite cursor

usage in a with statement [2]::

    with SQLiteCursor('test.db') as cur:
        print(cur.execute('select sqlite_version();').fetchall()[0][0])

[1] https://docs.python.org/3/library/stdtypes.html#context-manager-types
[2] https://docs.python.org/3/reference/compound_stmts.html#with

Signed-off-by: Markus Heiser <markus.heiser@darmarit.de>
This commit is contained in:
Markus Heiser 2021-05-20 17:34:46 +02:00
parent e0d64a0e2c
commit 10ea2eef3a
2 changed files with 36 additions and 29 deletions

View file

@ -6,9 +6,8 @@
"""
import sqlite3
from searx import logger
from searx.utils import SQLiteCursor
logger = logger.getChild('SQLite engine')
@ -19,32 +18,6 @@ limit = 10
paging = True
result_template = 'key-value.html'
class SQLiteDB:
"""
Implements a `Context Manager`_ for a SQLite.
usage::
with SQLiteDB('test.db') as cur:
print(cur.execute('select sqlite_version();').fetchall()[0][0])
.. _Context Manager: https://docs.python.org/3/library/stdtypes.html#context-manager-types
"""
def __init__(self, db):
self.database = db
self.connect = None
def __enter__(self):
self.connect = sqlite3.connect(self.database)
self.connect.row_factory = sqlite3.Row
return self.connect.cursor()
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.connect.commit()
self.connect.close()
def init(engine_settings):
if 'query_str' not in engine_settings:
raise ValueError('query_str cannot be empty')
@ -64,7 +37,7 @@ def search(query, params):
}
query_to_run = query_str + ' LIMIT :limit OFFSET :offset'
with SQLiteDB(database) as cur:
with SQLiteCursor(database) as cur:
cur.execute(query_to_run, query_params)
col_names = [cn[0] for cn in cur.description]

View file

@ -9,6 +9,7 @@
import sys
import re
import importlib
import sqlite3
from numbers import Number
from os.path import splitext, join
@ -612,3 +613,36 @@ def eval_xpath_getindex(elements, xpath_spec, index, default=NOTSET):
# to record xpath_spec
raise SearxEngineXPathException(xpath_spec, 'index ' + str(index) + ' not found')
return default
class SQLiteCursor:
"""Implements a `Context Manager`_ for a SQLite *cursor*
Returns :py:obj:`sqlite3.Cursor` on entering the runtime context of a `with
statement`_.
usage::
with SQLiteCursor('test.db') as cur:
print(cur.execute('select sqlite_version();').fetchall()[0][0])
.. _Context Manager: https://docs.python.org/3/library/stdtypes.html#context-manager-types
.. _with statement: https://docs.python.org/3/reference/compound_stmts.html#with
"""
def __init__(self, db):
self.database = db
self.connect = None
self.cursor = None
def __enter__(self):
self.connect = sqlite3.connect(self.database)
self.connect.row_factory = sqlite3.Row
self.cursor = self.connect.cursor()
return self.cursor
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.connect.commit()
self.cursor.close()
self.connect.close()