soorena62
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
from typing import Dict, Any
|
|
from radiuma_api.asset import Asset
|
|
from radiuma_api.io_port import InPort
|
|
from radiuma_api.io_port import OutPort
|
|
# Write Codes Here:
|
|
|
|
|
|
class ExecutionContext:
|
|
"""
|
|
Holds runtime execution data for a Workflow run.
|
|
"""
|
|
def __init__(self, execution_id: str | None = None):
|
|
self.execution_id = execution_id
|
|
self._data: Dict[OutPort, Asset] = {}
|
|
self._metadata: Dict[str, Any] = {}
|
|
|
|
# Output Data Recording
|
|
def put(self, output_port: OutPort, asset: Asset):
|
|
"""Register an asset produced by an OutputPort."""
|
|
self._data[output_port] = asset
|
|
|
|
# Getting Data For InputPort
|
|
def get(self, input_port: InPort):
|
|
"""Retrieve the asset connected to this InputPort."""
|
|
if input_port.connected_output is None:
|
|
return None
|
|
return self._data.get(input_port.connected_output)
|
|
|
|
# Check Data Availability
|
|
def has_data(self, input_port: InPort):
|
|
return self.get(input_port) is not None
|
|
|
|
# Metadata management
|
|
def set_meta(self, key: str, value: Any):
|
|
self._metadata[key] = value
|
|
|
|
def get_meta(self, key: str, default=None):
|
|
return self._metadata.get(key, default)
|
|
|
|
# Status View
|
|
def status_view(self):
|
|
"""Return a lightweight snapshot of current execution state."""
|
|
return {
|
|
"execution_id": self.execution_id,
|
|
"assets": {
|
|
f"{port.parent_task.name}.{port.name}": {
|
|
"contract": repr(port.contract),
|
|
"asset_type": asset.type_name
|
|
}
|
|
for port, asset in self._data.items()
|
|
},
|
|
"metadata": self._metadata.copy(),
|
|
}
|