| Server IP : 188.40.95.74 / Your IP : 216.73.216.142 Web Server : Apache System : Linux cp01.striminghost.net 3.10.0-1160.119.1.el7.tuxcare.els13.x86_64 #1 SMP Fri Nov 22 06:29:45 UTC 2024 x86_64 User : vlasotin ( 1054) PHP Version : 5.6.40 Disable Function : NONE MySQL : ON | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON 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,
)