X7ROOT File Manager
Current Path:
/opt/imunify360/venv/lib/python3.11/site-packages/playhouse
opt
/
imunify360
/
venv
/
lib
/
python3.11
/
site-packages
/
playhouse
/
ðŸ“
..
📄
__init__.py
(0 B)
ðŸ“
__pycache__
📄
apsw_ext.py
(4.82 KB)
📄
cockroachdb.py
(9 KB)
📄
dataset.py
(14.09 KB)
📄
db_url.py
(4.15 KB)
📄
fields.py
(1.66 KB)
📄
flask_utils.py
(8 KB)
📄
hybrid.py
(1.49 KB)
📄
kv.py
(5.48 KB)
📄
migrate.py
(30.11 KB)
📄
mysql_ext.py
(3.17 KB)
📄
pool.py
(11.21 KB)
📄
postgres_ext.py
(14.41 KB)
📄
psycopg3_ext.py
(1.15 KB)
📄
reflection.py
(30.2 KB)
📄
shortcuts.py
(11.25 KB)
📄
signals.py
(2.46 KB)
📄
sqlcipher_ext.py
(3.55 KB)
📄
sqlite_changelog.py
(4.68 KB)
📄
sqlite_ext.py
(45.65 KB)
📄
sqlite_udf.py
(13.34 KB)
📄
sqliteq.py
(9.75 KB)
📄
test_utils.py
(1.81 KB)
Editing: signals.py
""" Provide django-style hooks for model events. """ from peewee import Model as _Model class Signal(object): def __init__(self): self._flush() def _flush(self): self._receivers = set() self._receiver_list = [] def connect(self, receiver, name=None, sender=None): name = name or receiver.__name__ key = (name, sender) if key not in self._receivers: self._receivers.add(key) self._receiver_list.append((name, receiver, sender)) else: raise ValueError('receiver named %s (for sender=%s) already ' 'connected' % (name, sender or 'any')) def disconnect(self, receiver=None, name=None, sender=None): if receiver: name = name or receiver.__name__ if not name: raise ValueError('a receiver or a name must be provided') key = (name, sender) if key not in self._receivers: raise ValueError('receiver named %s for sender=%s not found.' % (name, sender or 'any')) self._receivers.remove(key) self._receiver_list = [(n, r, s) for n, r, s in self._receiver_list if n != name and s != sender] def __call__(self, name=None, sender=None): def decorator(fn): self.connect(fn, name, sender) return fn return decorator def send(self, instance, *args, **kwargs): sender = type(instance) responses = [] for n, r, s in self._receiver_list: if s is None or isinstance(instance, s): responses.append((r, r(sender, instance, *args, **kwargs))) return responses pre_save = Signal() post_save = Signal() pre_delete = Signal() post_delete = Signal() pre_init = Signal() class Model(_Model): def __init__(self, *args, **kwargs): super(Model, self).__init__(*args, **kwargs) pre_init.send(self) def save(self, *args, **kwargs): pk_value = self._pk if self._meta.primary_key else True created = kwargs.get('force_insert', False) or not bool(pk_value) pre_save.send(self, created=created) ret = super(Model, self).save(*args, **kwargs) post_save.send(self, created=created) return ret def delete_instance(self, *args, **kwargs): pre_delete.send(self) ret = super(Model, self).delete_instance(*args, **kwargs) post_delete.send(self) return ret
Upload File
Create Folder