FM / playhouse

/opt/alt/python38/lib64/python3.8/site-packages/playhouse/

/opt/alt/python38/lib64/python3.8/site-packages/playhouse

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
apsw_ext.py	5061	0644	edit dl rm
berkeleydb.py	4138	0644	edit dl rm
csv_loader.py	75	0644	edit dl rm
csv_utils.py	11608	0644	edit dl rm
dataset.py	10993	0644	edit dl rm
db_url.py	3893	0644	edit dl rm
djpeewee.py	7904	0644	edit dl rm
fields.py	11806	0644	edit dl rm
flask_utils.py	5663	0644	edit dl rm
gfk.py	6035	0644	edit dl rm
hybrid.py	1473	0644	edit dl rm
kv.py	4807	0644	edit dl rm
migrate.py	24057	0644	edit dl rm
pool.py	8201	0644	edit dl rm
postgres_ext.py	13448	0644	edit dl rm
read_slave.py	1449	0644	edit dl rm
reflection.py	21845	0644	edit dl rm
shortcuts.py	7203	0644	edit dl rm
signals.py	2249	0644	edit dl rm
sqlcipher_ext.py	4222	0644	edit dl rm
sqliteq.py	8539	0644	edit dl rm
sqlite_ext.py	35778	0644	edit dl rm
sqlite_udf.py	13576	0644	edit dl rm
test_utils.py	2799	0644	edit dl rm
_speedups.cpython-38-x86_64-linux-gnu.so	244264	0755	edit dl rm
_sqlite_ext.cpython-38-x86_64-linux-gnu.so	73688	0755	edit dl rm
_sqlite_udf.cpython-38-x86_64-linux-gnu.so	89768	0755	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /opt/alt/python38/lib64/python3.8/site-packages/playhouse/signals.py (2249B)

"""
Provide django-style hooks for model events.
"""
from peewee import Model as _Model

class Signal(object):
    def __init__(self):
        self._flush()

def connect(self, receiver, name=None, sender=None):
        name = name or receiver.__name__
        if name not in self._receivers:
            self._receivers[name] = (receiver, sender)
            self._receiver_list.append(name)
        else:
            raise ValueError('receiver named %s already connected' % name)

def disconnect(self, receiver=None, name=None):
        if receiver:
            name = receiver.__name__
        if name:
            del self._receivers[name]
            self._receiver_list.remove(name)
        else:
            raise ValueError('a receiver or a name must be provided')

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 name in self._receiver_list:
            r, s = self._receivers[name]
            if s is None or isinstance(instance, s):
                responses.append((r, r(sender, instance, *args, **kwargs)))
        return responses

def _flush(self):
        self._receivers = {}
        self._receiver_list = []

pre_save = Signal()
post_save = Signal()
pre_delete = Signal()
post_delete = Signal()
pre_init = Signal()
post_init = Signal()

class Model(_Model):
    def __init__(self, *args, **kwargs):
        super(Model, self).__init__(*args, **kwargs)
        pre_init.send(self)

def prepared(self):
        super(Model, self).prepared()
        post_init.send(self)

def save(self, *args, **kwargs):
        pk_value = self._get_pk_value()
        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

""" Provide django-style hooks for model events. """ from peewee import Model as _Model class Signal(object): def __init__(self): self._flush() def connect(self, receiver, name=None, sender=None): name = name or receiver.__name__ if name not in self._receivers: self._receivers[name] = (receiver, sender) self._receiver_list.append(name) else: raise ValueError('receiver named %s already connected' % name) def disconnect(self, receiver=None, name=None): if receiver: name = receiver.__name__ if name: del self._receivers[name] self._receiver_list.remove(name) else: raise ValueError('a receiver or a name must be provided') 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 name in self._receiver_list: r, s = self._receivers[name] if s is None or isinstance(instance, s): responses.append((r, r(sender, instance, *args, **kwargs))) return responses def _flush(self): self._receivers = {} self._receiver_list = [] pre_save = Signal() post_save = Signal() pre_delete = Signal() post_delete = Signal() pre_init = Signal() post_init = Signal() class Model(_Model): def __init__(self, *args, **kwargs): super(Model, self).__init__(*args, **kwargs) pre_init.send(self) def prepared(self): super(Model, self).prepared() post_init.send(self) def save(self, *args, **kwargs): pk_value = self._get_pk_value() 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