Source code for bw2data.sqlite

import json
import pickle

from peewee import BlobField, SqliteDatabase, TextField

[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): self._filepath = filepath self._tables = tables 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
[docs] def db(self): return self._database
[docs] def change_path(self, filepath): self.db.close() 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): print("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): data = json.loads(value) if isinstance(data, list): data = tuple(data) return data