searx.sqlitedb.DBSession Class Reference

Collaboration diagram for searx.sqlitedb.DBSession:

Public Member Functions
sqlite3.Connection	get_connect (cls, "SQLiteAppl" app)
	__init__ (self, "SQLiteAppl" app)
sqlite3.Connection	conn (self)
	__del__ (self)

Public Attributes
uuid.UUID	uuid = uuid.uuid4()
SQLiteAppl	app = app

Protected Attributes
sqlite3.Connection|None	_conn = None

Detailed Description

A *thead-local* DB session

Definition at line 39 of file sqlitedb.py.

Constructor & Destructor Documentation

__init__()

searx.sqlitedb.DBSession.__init__	(		self,
		"SQLiteAppl"	app )

Definition at line 56 of file sqlitedb.py.

    def __init__(self, app: "SQLiteAppl"):
        self.uuid: uuid.UUID = uuid.uuid4()
        self.app: SQLiteAppl = app
        self._conn: sqlite3.Connection | None = None
        # self.__del__ will be called, when thread ends
        if getattr(THREAD_LOCAL, "DBSession_map", None) is None:
            url_to_session: dict[str, DBSession] = {}
            THREAD_LOCAL.DBSession_map = url_to_session
        THREAD_LOCAL.DBSession_map[self.app.db_url] = self
 

__del__()

searx.sqlitedb.DBSession.__del__

(

self

)

Definition at line 77 of file sqlitedb.py.

    def __del__(self):
        try:
            if self._conn is not None:
                # HINT: Don't use Python's logging facility in a destructor, it
                # will produce error reports when python aborts the process or
                # thread, because at this point objects that the logging module
                # needs, do not exist anymore.
                # msg = f"DBSession: close [{self.uuid}] {self.app.__class__.__name__}({self.app.db_url})"
                # logger.debug(msg)
                self._conn.close()
        except Exception:  # pylint: disable=broad-exception-caught
            pass
 
 

References _conn.

Member Function Documentation

conn()

sqlite3.Connection searx.sqlitedb.DBSession.conn

(

self

)

Definition at line 67 of file sqlitedb.py.

    def conn(self) -> sqlite3.Connection:
        msg = f"[{threading.current_thread().ident}] DBSession: " f"{self.app.__class__.__name__}({self.app.db_url})"
        if self._conn is None:
            self._conn = self.app.connect()
            logger.debug("%s --> created new connection", msg)
        # else:
        #     logger.debug("%s --> already connected", msg)
 
        return self._conn
 

References _conn, and app.

get_connect()

sqlite3.Connection searx.sqlitedb.DBSession.get_connect	(		cls,
		"SQLiteAppl"	app )

Returns a thread local DB connection.  The connection is only
established once per thread.

Definition at line 43 of file sqlitedb.py.

    def get_connect(cls, app: "SQLiteAppl") -> sqlite3.Connection:
        """Returns a thread local DB connection.  The connection is only
        established once per thread.
        """
        if getattr(THREAD_LOCAL, "DBSession_map", None) is None:
            url_to_session: dict[str, DBSession] = {}
            THREAD_LOCAL.DBSession_map = url_to_session
 
        session: DBSession | None = THREAD_LOCAL.DBSession_map.get(app.db_url)
        if session is None:
            session = cls(app)
        return session.conn
 

Member Data Documentation

_conn

sqlite3.Connection | None searx.sqlitedb.DBSession._conn = None

protected

Definition at line 59 of file sqlitedb.py.

Referenced by __del__(), and conn().

app

SQLiteAppl searx.sqlitedb.DBSession.app = app

Definition at line 58 of file sqlitedb.py.

Referenced by conn().

uuid

uuid.UUID searx.sqlitedb.DBSession.uuid = uuid.uuid4()

Definition at line 57 of file sqlitedb.py.

---

The documentation for this class was generated from the following file:

• /home/andrew/Documents/code/public/searxng/searx/sqlitedb.py