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/test_utils.py (2799B)

from functools import wraps
import logging

from peewee import create_model_tables
from peewee import drop_model_tables

logger = logging.getLogger('peewee')

class test_database(object):
    def __init__(self, db, models, create_tables=True, drop_tables=True,
                 fail_silently=False):
        if not isinstance(models, (list, tuple, set)):
            raise ValueError('%r must be a list or tuple.' % models)
        self.db = db
        self.models = models
        self.create_tables = create_tables
        self.drop_tables = drop_tables
        self.fail_silently = fail_silently

def __enter__(self):
        self.orig = []
        for m in self.models:
            self.orig.append(m._meta.database)
            m._meta.database = self.db
        if self.create_tables:
            create_model_tables(self.models, fail_silently=self.fail_silently)

def __exit__(self, exc_type, exc_val, exc_tb):
        if self.create_tables and self.drop_tables:
            drop_model_tables(self.models, fail_silently=self.fail_silently)
        for i, m in enumerate(self.models):
            m._meta.database = self.orig[i]

class _QueryLogHandler(logging.Handler):
    def __init__(self, *args, **kwargs):
        self.queries = []
        logging.Handler.__init__(self, *args, **kwargs)

def emit(self, record):
        self.queries.append(record)

class count_queries(object):
    def __init__(self, only_select=False):
        self.only_select = only_select
        self.count = 0

def get_queries(self):
        return self._handler.queries

def __enter__(self):
        self._handler = _QueryLogHandler()
        logger.setLevel(logging.DEBUG)
        logger.addHandler(self._handler)
        return self

def __exit__(self, exc_type, exc_val, exc_tb):
        logger.removeHandler(self._handler)
        if self.only_select:
            self.count = len([q for q in self._handler.queries
                              if q.msg[0].startswith('SELECT ')])
        else:
            self.count = len(self._handler.queries)

class assert_query_count(count_queries):
    def __init__(self, expected, only_select=False):
        super(assert_query_count, self).__init__(only_select=only_select)
        self.expected = expected

def __call__(self, f):
        @wraps(f)
        def decorated(*args, **kwds):
            with self:
                ret = f(*args, **kwds)

self._assert_count()
            return ret

return decorated

def _assert_count(self):
        error_msg = '%s != %s' % (self.count, self.expected)
        assert self.count == self.expected, error_msg

def __exit__(self, exc_type, exc_val, exc_tb):
        super(assert_query_count, self).__exit__(exc_type, exc_val, exc_tb)
        self._assert_count()

from functools import wraps import logging from peewee import create_model_tables from peewee import drop_model_tables logger = logging.getLogger('peewee') class test_database(object): def __init__(self, db, models, create_tables=True, drop_tables=True, fail_silently=False): if not isinstance(models, (list, tuple, set)): raise ValueError('%r must be a list or tuple.' % models) self.db = db self.models = models self.create_tables = create_tables self.drop_tables = drop_tables self.fail_silently = fail_silently def __enter__(self): self.orig = [] for m in self.models: self.orig.append(m._meta.database) m._meta.database = self.db if self.create_tables: create_model_tables(self.models, fail_silently=self.fail_silently) def __exit__(self, exc_type, exc_val, exc_tb): if self.create_tables and self.drop_tables: drop_model_tables(self.models, fail_silently=self.fail_silently) for i, m in enumerate(self.models): m._meta.database = self.orig[i] class _QueryLogHandler(logging.Handler): def __init__(self, *args, **kwargs): self.queries = [] logging.Handler.__init__(self, *args, **kwargs) def emit(self, record): self.queries.append(record) class count_queries(object): def __init__(self, only_select=False): self.only_select = only_select self.count = 0 def get_queries(self): return self._handler.queries def __enter__(self): self._handler = _QueryLogHandler() logger.setLevel(logging.DEBUG) logger.addHandler(self._handler) return self def __exit__(self, exc_type, exc_val, exc_tb): logger.removeHandler(self._handler) if self.only_select: self.count = len([q for q in self._handler.queries if q.msg[0].startswith('SELECT ')]) else: self.count = len(self._handler.queries) class assert_query_count(count_queries): def __init__(self, expected, only_select=False): super(assert_query_count, self).__init__(only_select=only_select) self.expected = expected def __call__(self, f): @wraps(f) def decorated(*args, **kwds): with self: ret = f(*args, **kwds) self._assert_count() return ret return decorated def _assert_count(self): error_msg = '%s != %s' % (self.count, self.expected) assert self.count == self.expected, error_msg def __exit__(self, exc_type, exc_val, exc_tb): super(assert_query_count, self).__exit__(exc_type, exc_val, exc_tb) self._assert_count()