Hackfut Security File Manager
Current Path:
/opt/alt/python38/lib64/python3.8/site-packages/playhouse
opt
/
alt
/
python38
/
lib64
/
python3.8
/
site-packages
/
playhouse
/
📁
..
📄
__init__.py
(0 B)
📁
__pycache__
📄
_speedups.cpython-38-x86_64-linux-gnu.so
(235.1 KB)
📄
_sqlite_ext.cpython-38-x86_64-linux-gnu.so
(70.62 KB)
📄
_sqlite_udf.cpython-38-x86_64-linux-gnu.so
(86.17 KB)
📄
apsw_ext.py
(4.94 KB)
📄
berkeleydb.py
(4.04 KB)
📄
csv_loader.py
(75 B)
📄
csv_utils.py
(11.34 KB)
📄
dataset.py
(10.74 KB)
📄
db_url.py
(3.8 KB)
📄
djpeewee.py
(7.72 KB)
📄
fields.py
(11.53 KB)
📄
flask_utils.py
(5.53 KB)
📄
gfk.py
(5.89 KB)
📄
hybrid.py
(1.44 KB)
📄
kv.py
(4.69 KB)
📄
migrate.py
(23.49 KB)
📄
pool.py
(8.01 KB)
📄
postgres_ext.py
(13.13 KB)
📄
read_slave.py
(1.42 KB)
📄
reflection.py
(21.33 KB)
📄
shortcuts.py
(7.03 KB)
📄
signals.py
(2.2 KB)
📄
sqlcipher_ext.py
(4.12 KB)
📄
sqlite_ext.py
(34.94 KB)
📄
sqlite_udf.py
(13.26 KB)
📄
sqliteq.py
(8.34 KB)
📄
test_utils.py
(2.73 KB)
Editing: apsw_ext.py
""" Peewee integration with APSW, "another python sqlite wrapper". Project page: https://rogerbinns.github.io/apsw/ APSW is a really neat library that provides a thin wrapper on top of SQLite's C interface. Here are just a few reasons to use APSW, taken from the documentation: * APSW gives all functionality of SQLite, including virtual tables, virtual file system, blob i/o, backups and file control. * Connections can be shared across threads without any additional locking. * Transactions are managed explicitly by your code. * APSW can handle nested transactions. * Unicode is handled correctly. * APSW is faster. """ import apsw from peewee import * from peewee import _sqlite_date_part from peewee import _sqlite_date_trunc from peewee import _sqlite_regexp from peewee import BooleanField as _BooleanField from peewee import DateField as _DateField from peewee import DateTimeField as _DateTimeField from peewee import DecimalField as _DecimalField from peewee import logger from peewee import savepoint from peewee import TimeField as _TimeField from peewee import transaction as _transaction from playhouse.sqlite_ext import SqliteExtDatabase from playhouse.sqlite_ext import VirtualCharField from playhouse.sqlite_ext import VirtualField from playhouse.sqlite_ext import VirtualFloatField from playhouse.sqlite_ext import VirtualIntegerField from playhouse.sqlite_ext import VirtualModel class transaction(_transaction): def __init__(self, db, lock_type='deferred'): self.db = db self.lock_type = lock_type def _begin(self): self.db.begin(self.lock_type) class APSWDatabase(SqliteExtDatabase): def __init__(self, database, timeout=None, **kwargs): self.timeout = timeout self._modules = {} super(APSWDatabase, self).__init__(database, **kwargs) def register_module(self, mod_name, mod_inst): self._modules[mod_name] = mod_inst def unregister_module(self, mod_name): del(self._modules[mod_name]) def _connect(self, database, **kwargs): conn = apsw.Connection(database, **kwargs) if self.timeout is not None: conn.setbusytimeout(self.timeout) try: self._add_conn_hooks(conn) except: conn.close() raise return conn def _add_conn_hooks(self, conn): super(APSWDatabase, self)._add_conn_hooks(conn) self._load_modules(conn) # APSW-only. def _load_modules(self, conn): for mod_name, mod_inst in self._modules.items(): conn.createmodule(mod_name, mod_inst) return conn def _load_aggregates(self, conn): for name, (klass, num_params) in self._aggregates.items(): def make_aggregate(): instance = klass() return (instance, instance.step, instance.finalize) conn.createaggregatefunction(name, make_aggregate) def _load_collations(self, conn): for name, fn in self._collations.items(): conn.createcollation(name, fn) def _load_functions(self, conn): for name, (fn, num_params) in self._functions.items(): conn.createscalarfunction(name, fn, num_params) def _load_extensions(self, conn): conn.enableloadextension(True) for extension in self._extensions: conn.loadextension(extension) def load_extension(self, extension): self._extensions.add(extension) if not self.is_closed(): conn = self.get_conn() conn.enableloadextension(True) conn.loadextension(extension) def _execute_sql(self, cursor, sql, params): cursor.execute(sql, params or ()) return cursor def execute_sql(self, sql, params=None, require_commit=True): logger.debug((sql, params)) with self.exception_wrapper(): cursor = self.get_cursor() self._execute_sql(cursor, sql, params) return cursor def last_insert_id(self, cursor, model): if model._meta.auto_increment: return cursor.getconnection().last_insert_rowid() def rows_affected(self, cursor): return cursor.getconnection().changes() def begin(self, lock_type='deferred'): self.get_cursor().execute('begin %s;' % lock_type) def commit(self): self.get_cursor().execute('commit;') def rollback(self): self.get_cursor().execute('rollback;') def transaction(self, lock_type='deferred'): return transaction(self, lock_type) def savepoint(self, sid=None): return savepoint(self, sid) def nh(s, v): if v is not None: return str(v) class BooleanField(_BooleanField): def db_value(self, v): v = super(BooleanField, self).db_value(v) if v is not None: return v and 1 or 0 class DateField(_DateField): db_value = nh class TimeField(_TimeField): db_value = nh class DateTimeField(_DateTimeField): db_value = nh class DecimalField(_DecimalField): db_value = nh
Upload File
Create Folder