| Server IP : 68.65.120.201 / Your IP : 216.73.216.221 Web Server : LiteSpeed System : Linux server179.web-hosting.com 4.18.0-513.18.1.lve.el8.x86_64 #1 SMP Thu Feb 22 12:55:50 UTC 2024 x86_64 User : taxhyuvu ( 2294) PHP Version : 8.1.33 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/model/ |
Upload File : |
import logging
import threading
from playhouse.sqlite_ext import SqliteExtDatabase
class OverridingReset(Exception):
"""
Overriding reset could be a signal of logic error
thus need to be explicitly handled in all places where
this exception is expected to occur.
"""
pass
logger = logging.getLogger(__name__)
_thread_local_storage = threading.local()
class SqliteDatabaseWrapper(SqliteExtDatabase):
def execute_sql(self, *args, **kwargs):
_validate(*args, **kwargs)
return super().execute_sql(*args, **kwargs)
def reset(new_value=None):
if hasattr(_thread_local_storage, "thread_ident_memo"):
raise OverridingReset()
_thread_local_storage.thread_ident_memo = (
new_value or threading.get_ident()
)
def _validate(*args, **kwargs):
thread_ident_memo = getattr(
_thread_local_storage, "thread_ident_memo", None
)
if thread_ident_memo is None:
logger.error("wrong thread or _validate() was not preceded by reset()")
elif thread_ident_memo != threading.get_ident():
logger.error(
"thread_ident_memo check failed [%r != %r]\n"
"context:\nargs: %s\nkwargs: %s",
thread_ident_memo,
threading.get_ident(),
args,
kwargs,
)