Source code for bw2data.sqlite
import json
import pickle
from peewee import BlobField, SqliteDatabase, TextField
from bw2data.logs import stdout_feedback_logger
[docs]
class PickleField(BlobField):
[docs]
def db_value(self, value):
return super(PickleField, self).db_value(pickle.dumps(value, protocol=4))
[docs]
def python_value(self, value):
return pickle.loads(bytes(value))
[docs]
class SubstitutableDatabase:
def __init__(self, filepath, tables):
[docs]
self._filepath = filepath
[docs]
self._database = self._create_database()
[docs]
def _create_database(self):
db = SqliteDatabase(self._filepath)
for model in self._tables:
model.bind(db, bind_refs=False, bind_backrefs=False)
db.connect()
db.create_tables(self._tables)
return db
@property
[docs]
def db(self):
return self._database
[docs]
def change_path(self, filepath):
import gc
old_db = self._database
old_db.close()
del old_db
gc.collect()
self._filepath = filepath
self._database = self._create_database()
[docs]
def atomic(self):
return self.db.atomic()
[docs]
def execute_sql(self, *args, **kwargs):
return self.db.execute_sql(*args, **kwargs)
[docs]
def transaction(self):
return self.db.transaction()
[docs]
def vacuum(self):
stdout_feedback_logger.info("Vacuuming database ")
self.execute_sql("VACUUM;")
[docs]
class JSONField(TextField):
"""Simpler JSON field that doesn't support advanced querying and is human-readable"""
[docs]
def db_value(self, value):
return super().db_value(
json.dumps(
value,
ensure_ascii=False,
indent=2,
default=lambda x: x.isoformat() if hasattr(x, "isoformat") else x,
)
)
[docs]
def python_value(self, value):
return json.loads(value)
[docs]
class TupleJSONField(JSONField):
[docs]
def python_value(self, value):
if value is None:
return None
data = json.loads(value)
if isinstance(data, list):
data = tuple(data)
return data
