vikarti.anatra/Skyvern
1
0
Fork 0
mirror of https://github.com/Skyvern-AI/skyvern.git synced 2025-09-09 22:06:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

forloop metadata variables (#1334)

Browse source
This commit is contained in:
LawyZheng 2024-12-06 11:35:32 +08:00 • committed by GitHub
parent 01e9678d27
commit db5b9d1dbd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 34 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

15
skyvern/forge/sdk/workflow/context_manager.py
View file

@ -25,6 +25,9 @@ if TYPE_CHECKING:
LOG = structlog.get_logger()
BlockMetadata = dict[str, str | int | float | bool | dict | list]
class WorkflowRunContext:
parameters: dict[str, PARAMETER_TYPE]
values: dict[str, Any]
@ -36,9 +39,12 @@ class WorkflowRunContext:
workflow_output_parameters: list[OutputParameter],
context_parameters: list[ContextParameter],
) -> None:
# key is label name
self.blocks_metadata: dict[str, BlockMetadata] = {}
self.parameters = {}
self.values = {}
self.secrets = {}
for parameter, run_parameter in workflow_parameter_tuples:
if parameter.key in self.parameters:
prev_value = self.parameters[parameter.key]
@ -81,6 +87,15 @@ class WorkflowRunContext:
def set_value(self, key: str, value: Any) -> None:
self.values[key] = value
def update_block_metadata(self, label: str, metadata: BlockMetadata) -> None:
if label in self.blocks_metadata:
self.blocks_metadata[label].update(metadata)
return
self.blocks_metadata[label] = metadata
def get_block_metadata(self, label: str) -> BlockMetadata:
return self.blocks_metadata.get(label, BlockMetadata())
def get_original_secret_value_or_none(self, secret_id_or_value: Any) -> Any:
"""
Get the original secret value from the secrets dict. If the secret id is not found, return None.