|
40 | 40 |
|
41 | 41 | LEDGER_SCHEMA = "datalab-provenance-ledger" |
42 | 42 | LEDGER_SCHEMA_VERSION = 1 |
43 | | -#: Origins written today; ``recipe``, ``legacy_metadata`` and ``opaque`` are |
44 | | -#: reserved and accepted when reading. |
| 43 | +#: Origins written by :meth:`Ledger.record_activity`; ``legacy_metadata`` is |
| 44 | +#: written by :meth:`Ledger.record_legacy_activity`; ``recipe`` and ``opaque`` |
| 45 | +#: are reserved and accepted when reading. |
45 | 46 | ORIGINS = ("ordinary", "recompute_in_place", "history_replay") |
46 | 47 |
|
47 | 48 |
|
@@ -106,17 +107,26 @@ def validate_ledger(data: Any) -> None: |
106 | 107 | if act_id in activity_ids: |
107 | 108 | problems.append(f"{where}: duplicate activity_id") |
108 | 109 | activity_ids.add(act_id) |
| 110 | + legacy = activity["origin"] == "legacy_metadata" |
109 | 111 | try: |
110 | | - validate_call(activity["call"]) |
| 112 | + validate_call(activity["call"], allow_object_bindings=legacy) |
111 | 113 | except CallError as exc: |
112 | 114 | problems.append(f"{where}: {exc}") |
113 | 115 | continue |
114 | 116 | for item in activity["call"]["inputs"]: |
115 | | - if item["binding"]["state_id"] not in states: |
| 117 | + state_id = item["binding"].get("state_id") |
| 118 | + if state_id is not None and state_id not in states: |
116 | 119 | problems.append(f"{where}: unknown input state") |
117 | | - if is_opaque(activity["call"]) and activity["implementation"] is None: |
| 120 | + if legacy and not is_opaque(activity["call"]): |
| 121 | + problems.append(f"{where}: a legacy activity is never qualified") |
| 122 | + if ( |
| 123 | + is_opaque(activity["call"]) |
| 124 | + and activity["implementation"] is None |
| 125 | + and not legacy |
| 126 | + ): |
118 | 127 | problems.append(f"{where}: an opaque call needs an implementation") |
119 | | - if activity["environment_id"] not in environments: |
| 128 | + env_id = activity["environment_id"] |
| 129 | + if env_id is not None and env_id not in environments: |
120 | 130 | problems.append(f"{where}: unknown environment") |
121 | 131 | for output in activity["outputs"]: |
122 | 132 | state_id = output.get("state_id") |
@@ -385,3 +395,71 @@ def record_activity( |
385 | 395 | self._activity_index[activity_id] = len(self._data["activities"]) |
386 | 396 | self._data["activities"].append(activity) |
387 | 397 | return activity |
| 398 | + |
| 399 | + def record_legacy_activity( |
| 400 | + self, |
| 401 | + *, |
| 402 | + inputs: Iterable[tuple[str, str]], |
| 403 | + outputs: Iterable[tuple[str, str, dict[str, Any]]], |
| 404 | + legacy: dict[str, Any], |
| 405 | + limits: Iterable[str] = (), |
| 406 | + ) -> dict[str, Any]: |
| 407 | + """Append an activity rebuilt from processing metadata of an old workspace. |
| 408 | +
|
| 409 | + What the metadata does not say stays unknown: inputs are bound to objects |
| 410 | + (their state at execution time is unknown), the call is opaque with no |
| 411 | + parameters, and environment, edition, implementation and times are null. |
| 412 | + Such an activity is never replayed. |
| 413 | +
|
| 414 | + Args: |
| 415 | + inputs: Ordered ``(role, object_uuid)`` pairs. |
| 416 | + outputs: Ordered ``(role, object_uuid, facts)`` triples, observed when |
| 417 | + the workspace was opened. |
| 418 | + legacy: ``{"key", "object_uuid", "record", "saved_by_version", |
| 419 | + "issues"}`` (see :mod:`datalab_capsule.legacy`). |
| 420 | + limits: Reasons why the activity is not replayable. |
| 421 | +
|
| 422 | + Returns: |
| 423 | + The appended activity. |
| 424 | +
|
| 425 | + Raises: |
| 426 | + LedgerError: If the activity is inconsistent. |
| 427 | + """ |
| 428 | + call = { |
| 429 | + "operation": None, |
| 430 | + "parameters": None, |
| 431 | + "inputs": [ |
| 432 | + {"role": role, "binding": {"object_uuids": [obj_uuid]}} |
| 433 | + for role, obj_uuid in inputs |
| 434 | + ], |
| 435 | + } |
| 436 | + try: |
| 437 | + validate_call(call, allow_object_bindings=True) |
| 438 | + except CallError as exc: |
| 439 | + raise LedgerError(str(exc)) from exc |
| 440 | + activity_id = str(uuid.uuid4()) |
| 441 | + activity = { |
| 442 | + "activity_id": activity_id, |
| 443 | + "command_id": None, |
| 444 | + "call": call, |
| 445 | + "implementation": None, |
| 446 | + "outputs": [ |
| 447 | + { |
| 448 | + "role": role, |
| 449 | + "state_id": self._new_state(obj_uuid, facts, activity_id), |
| 450 | + } |
| 451 | + for role, obj_uuid, facts in outputs |
| 452 | + ], |
| 453 | + "context": {}, |
| 454 | + "limits": list(limits), |
| 455 | + "environment_id": None, |
| 456 | + "edition": None, |
| 457 | + "origin": "legacy_metadata", |
| 458 | + "status": "completed", |
| 459 | + "started_at": None, |
| 460 | + "finished_at": None, |
| 461 | + "legacy": copy.deepcopy(legacy), |
| 462 | + } |
| 463 | + self._activity_index[activity_id] = len(self._data["activities"]) |
| 464 | + self._data["activities"].append(activity) |
| 465 | + return activity |
0 commit comments