Source code for bw2data.backends.schema
from peewee import DoesNotExist, TextField
from bw2data.errors import UnknownObject
from bw2data.signals import (
on_activity_code_change,
on_activity_database_change,
on_database_delete,
on_database_reset,
on_database_write,
project_changed,
signaleddataset_on_delete,
)
from bw2data.snowflake_ids import SnowflakeIDBaseClass
from bw2data.sqlite import PickleField
[docs]
class ActivityDataset(SnowflakeIDBaseClass):
[docs]
data = PickleField() # Canonical, except for other C fields
[docs]
code = TextField() # Canonical
[docs]
database = TextField() # Canonical
[docs]
location = TextField(null=True) # Reset from `data`
[docs]
name = TextField(null=True) # Reset from `data`
[docs]
product = TextField(null=True) # Reset from `data`
[docs]
type = TextField(null=True) # Reset from `data`
@property
[docs]
def key(self):
return (self.database, self.code)
[docs]
class ExchangeDataset(SnowflakeIDBaseClass):
[docs]
data = PickleField() # Canonical, except for other C fields
[docs]
output_code = TextField() # Canonical
[docs]
output_database = TextField() # Canonical
[docs]
type = TextField() # Reset from `data`
[docs]
_get_id_cache: dict = {}
[docs]
def get_id(key):
if isinstance(key, int):
try:
ActivityDataset.get(ActivityDataset.id == key)
except DoesNotExist:
raise UnknownObject
return key
else:
cache_key = (key[0], key[1])
if cache_key in _get_id_cache:
return _get_id_cache[cache_key]
try:
result = ActivityDataset.get(
ActivityDataset.database == key[0], ActivityDataset.code == key[1]
).id
_get_id_cache[cache_key] = result
return result
except DoesNotExist:
raise UnknownObject
[docs]
def _clear_get_id_cache(sender, **kwargs):
_get_id_cache.clear()
[docs]
def _remove_database_from_get_id_cache(sender, name: str, **kwargs):
for k in [k for k in _get_id_cache if k[0] == name]:
del _get_id_cache[k]
[docs]
def _remove_activity_from_get_id_cache(sender, old=None, **kwargs):
if isinstance(old, ActivityDataset):
_get_id_cache.pop((old.database, old.code), None)
[docs]
def _remove_changed_activity_key_from_get_id_cache(sender, old=None, **kwargs):
if old is not None:
for k in [k for k, v in _get_id_cache.items() if v == old["id"]]:
del _get_id_cache[k]
project_changed.connect(_clear_get_id_cache)
on_database_delete.connect(_remove_database_from_get_id_cache)
on_database_reset.connect(_remove_database_from_get_id_cache)
on_database_write.connect(_remove_database_from_get_id_cache)
signaleddataset_on_delete.connect(_remove_activity_from_get_id_cache)
on_activity_database_change.connect(_remove_changed_activity_key_from_get_id_cache)
on_activity_code_change.connect(_remove_changed_activity_key_from_get_id_cache)
