| 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 time
from peewee import CharField, IntegerField, CompositeKey
from defence360agent.contracts.config import IContactMessageType
from defence360agent.model import Model, instance
from defence360agent.utils.common import DAY, WEEK
THROTTLING_PERIOD = {
IContactMessageType.MALWARE_FOUND: DAY,
IContactMessageType.SCAN_NOT_SCHEDULED: WEEK,
}
class IContactThrottle(Model):
class Meta:
database = instance.db
db_table = "icontact_throttle"
primary_key = CompositeKey("message_type", "user")
message_type = CharField()
user = CharField(null=True)
#: The last time we sent a notification about :attr:`message_type`
timestamp = IntegerField(default=0)
@classmethod
def may_be_notified(cls, message_type, period_limit, user=None):
obj, _ = cls.get_or_create(message_type=message_type, user=user)
return (time.time() - obj.timestamp) > period_limit
@classmethod
def refresh(cls, message_type, user=None):
cls.update(timestamp=time.time()).where(
cls.message_type == message_type,
cls.user.is_null(True) if user is None else cls.user == user,
).execute()