bw2data.revisions#

Attributes#

Classes#

Delta

The difference between two versions of an object.

JSONEncoder

Extensible JSON <https://json.org> encoder for Python data structures.

RevisionGraph

Graph of revisions, edges are based on metadata.parent_revision.

RevisionedActivityParameter

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

RevisionedDatabase

RevisionedDatabaseParameter

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

RevisionedEdge

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

RevisionedGroup

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

RevisionedNode

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

RevisionedORMProxy

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

RevisionedParameter

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

RevisionedParameterizedExchange

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

RevisionedProjectParameter

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

Functions#

generate_metadata(→ dict[str, Any])

generate_revision(→ dict)

Module Contents#

class bw2data.revisions.Delta(delta: deepdiff.Delta | dict | None, obj_type: str | None = None, obj_id: int | str | None = None, change_type: str | None = None)[source]#

The difference between two versions of an object.

Can be serialized, transferred, and applied to the same previous version to change it to the new state.

Private, exists only for type-checking.

Use one of the class-method constructors to create objects.

classmethod _direct_lci_node_difference(old: dict, new: dict, change_type: str) Self[source]#
classmethod activity_code_change(old: dict, new: dict) Self[source]#

Special handling to change the database attribute of an activity node.

classmethod activity_database_change(old: dict, new: dict) Self[source]#

Special handling to change the database attribute of an activity node.

apply(obj)[source]#
classmethod database_metadata_change(old: dict, new: dict) Self | None[source]#

Special handling to change the database attribute of an activity node.

classmethod from_dict(d: dict) Self[source]#
classmethod from_difference(obj_type: str, obj_id: int | str | None, change_type: str, diff: deepdiff.DeepDiff) Self[source]#
classmethod generate(old: Any | None, new: Any | None, operation: str | None = None) Self[source]#

Generates a patch object from one version of an object to another.

Both old and new should be instances of bw2data.backends.schema.SignaledDataset.

old can be None if an object is being created.

new can be None is an object is being deleted.

Raises IncompatibleClasses is old and new have different classes.

change_type[source]#
delta[source]#
id[source]#
type[source]#
class bw2data.revisions.JSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]#

Bases: json.JSONEncoder

Extensible JSON <https://json.org> encoder for Python data structures.

Supports the following objects and types by default:

Python

JSON

dict

object

list, tuple

array

str

string

int, float

number

True

true

False

false

None

null

To extend this to recognize other objects, subclass and implement a .default() method with another method that returns a serializable object for o if possible, otherwise it should call the superclass implementation (to raise TypeError).

Constructor for JSONEncoder, with sensible defaults.

If skipkeys is false, then it is a TypeError to attempt encoding of keys that are not str, int, float or None. If skipkeys is True, such items are simply skipped.

If ensure_ascii is true, the output is guaranteed to be str objects with all incoming non-ASCII characters escaped. If ensure_ascii is false, the output can contain non-ASCII characters.

If check_circular is true, then lists, dicts, and custom encoded objects will be checked for circular references during encoding to prevent an infinite recursion (which would cause an RecursionError). Otherwise, no such check takes place.

If allow_nan is true, then NaN, Infinity, and -Infinity will be encoded as such. This behavior is not JSON specification compliant, but is consistent with most JavaScript based encoders and decoders. Otherwise, it will be a ValueError to encode such floats.

If sort_keys is true, then the output of dictionaries will be sorted by key; this is useful for regression tests to ensure that JSON serializations can be compared on a day-to-day basis.

If indent is a non-negative integer, then JSON array elements and object members will be pretty-printed with that indent level. An indent level of 0 will only insert newlines. None is the most compact representation.

If specified, separators should be an (item_separator, key_separator) tuple. The default is (’, ‘, ‘: ‘) if indent is None and (‘,’, ‘: ‘) otherwise. To get the most compact JSON representation, you should specify (‘,’, ‘:’) to eliminate whitespace.

If specified, default is a function that gets called for objects that can’t otherwise be serialized. It should return a JSON encodable version of the object or raise a TypeError.

default(obj)[source]#

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return super().default(o)
class bw2data.revisions.RevisionGraph(head: int, revisions: Sequence[dict])[source]#

Graph of revisions, edges are based on metadata.parent_revision.

class Iterator(g: RevisionGraph)[source]#

Helper class implementing iteration from child to parent.

head[source]#
id_map[source]#
head[source]#
id_map[source]#
revisions[source]#
class bw2data.revisions.RevisionedActivityParameter[source]#

Bases: RevisionedParameter

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

We need a separate class because we apply the changes to ActivityDataset, but need to save Node (and similar for edges).

classmethod activity_parameter_recalculate(revision_data: dict) None[source]#
classmethod activity_parameter_recalculate_exchanges(revision_data: dict) None[source]#
classmethod activity_parameter_update_formula_activity_parameter_name(revision_data: dict) None[source]#
classmethod activity_parameter_update_formula_database_parameter_name(revision_data: dict) None[source]#
classmethod activity_parameter_update_formula_project_parameter_name(revision_data: dict) None[source]#
KEYS = ('id', 'group', 'database', 'code', 'name', 'formula', 'amount', 'data')[source]#
ORM_CLASS[source]#
class bw2data.revisions.RevisionedDatabase[source]#
classmethod handle(revision_data: dict) None[source]#
class bw2data.revisions.RevisionedDatabaseParameter[source]#

Bases: RevisionedParameter

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

We need a separate class because we apply the changes to ActivityDataset, but need to save Node (and similar for edges).

classmethod database_parameter_recalculate(revision_data: dict) None[source]#
classmethod database_parameter_update_formula_database_parameter_name(revision_data: dict) None[source]#
classmethod database_parameter_update_formula_project_parameter_name(revision_data: dict) None[source]#
KEYS = ('id', 'database', 'name', 'formula', 'amount', 'data')[source]#
ORM_CLASS[source]#
class bw2data.revisions.RevisionedEdge[source]#

Bases: RevisionedORMProxy

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

We need a separate class because we apply the changes to ActivityDataset, but need to save Node (and similar for edges).

classmethod orm_as_dict(orm_object: bw2data.backends.proxies.Exchange.ORMDataset) dict[source]#
ORM_CLASS[source]#
PROXY_CLASS[source]#
class bw2data.revisions.RevisionedGroup[source]#

Bases: RevisionedParameter

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

We need a separate class because we apply the changes to ActivityDataset, but need to save Node (and similar for edges).

KEYS = ('id', 'name', 'order')[source]#
ORM_CLASS[source]#
class bw2data.revisions.RevisionedNode[source]#

Bases: RevisionedORMProxy

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

We need a separate class because we apply the changes to ActivityDataset, but need to save Node (and similar for edges).

classmethod activity_code_change(revision_data: dict) None[source]#

Special handling for changing activity code attributes

classmethod activity_database_change(revision_data: dict) None[source]#

Special handling for changing activity database attributes

classmethod orm_as_dict(orm_object: bw2data.backends.proxies.Activity.ORMDataset) dict[source]#
classmethod prepare_data_dict_for_orm_class(data: dict) dict[source]#
ORM_CLASS[source]#
PROXY_CLASS[source]#
class bw2data.revisions.RevisionedORMProxy[source]#

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

We need a separate class because we apply the changes to ActivityDataset, but need to save Node (and similar for edges).

classmethod create(revision_data: dict) None[source]#
classmethod current_state_as_dict(obj: bw2data.signals.SignaledDataset) dict[source]#
classmethod delete(revision_data: dict) None[source]#
classmethod handle(revision_data: dict) None[source]#
classmethod prepare_data_dict_for_orm_class(data: dict) dict[source]#
classmethod previous_state_as_dict(revision_data: dict) dict[source]#
classmethod update(revision_data: dict) None[source]#
class bw2data.revisions.RevisionedParameter[source]#

Bases: RevisionedORMProxy

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

We need a separate class because we apply the changes to ActivityDataset, but need to save Node (and similar for edges).

classmethod _state_as_dict(obj: bw2data.parameters.ParameterBase) dict[source]#
classmethod _unwrap_diff_dict(data: dict) dict[source]#
classmethod create(revision_data: dict) None[source]#
classmethod current_state_as_dict(obj: bw2data.parameters.ParameterBase) dict[source]#
classmethod delete(revision_data: dict) None[source]#
classmethod previous_state_as_dict(revision_data: dict) dict[source]#
classmethod update(revision_data: dict) None[source]#
class bw2data.revisions.RevisionedParameterizedExchange[source]#

Bases: RevisionedParameter

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

We need a separate class because we apply the changes to ActivityDataset, but need to save Node (and similar for edges).

KEYS = ('id', 'group', 'formula', 'exchange')[source]#
ORM_CLASS[source]#
class bw2data.revisions.RevisionedProjectParameter[source]#

Bases: RevisionedParameter

Class that encapsulates logic around applying revisions. Used for Activity and Exchange.

We need a separate class because we apply the changes to ActivityDataset, but need to save Node (and similar for edges).

classmethod project_parameter_recalculate(revision_data: dict) None[source]#
classmethod project_parameter_update_formula_parameter_name(revision_data: dict) None[source]#
KEYS = ('id', 'name', 'formula', 'amount', 'data')[source]#
ORM_CLASS[source]#
bw2data.revisions.generate_metadata(metadata: dict[str, Any] | None = None, parent_revision: int | None = None, revision: int | None = None) dict[str, Any][source]#
bw2data.revisions.generate_revision(metadata: dict, delta: Sequence[Delta]) dict[source]#
bw2data.revisions.REVISIONED_LABEL_AS_OBJECT[source]#
bw2data.revisions.REVISIONS_OBJECT_AS_LABEL[source]#
bw2data.revisions.SIGNALLEDOBJECT_TO_LABEL[source]#
bw2data.revisions.generate_delta[source]#