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

workflow parameter validation (#1028)

Browse source
This commit is contained in:
Shuchang Zheng 2024-10-22 17:36:25 -07:00 • committed by GitHub
parent 7cba401e2e
commit 0e3da8d1d3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 46 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

38
skyvern/forge/sdk/workflow/models/parameter.py
View file

@ -2,10 +2,12 @@ import abc
import json
from datetime import datetime
from enum import StrEnum
from typing import Annotated, Literal, Union
from typing import Annotated, Any, Literal, Union
from pydantic import BaseModel, ConfigDict, Field
from skyvern.exceptions import InvalidWorkflowParameter
class ParameterType(StrEnum):
WORKFLOW = "workflow"
@ -114,21 +116,29 @@ class WorkflowParameterType(StrEnum):
JSON = "json"
FILE_URL = "file_url"
def convert_value(self, value: str | None) -> str | int | float | bool | dict | list | None:
def convert_value(self, value: Any) -> str | int | float | bool | dict | list | None:
if value is None:
return None
if self == WorkflowParameterType.STRING:
return value
elif self == WorkflowParameterType.INTEGER:
return int(value)
elif self == WorkflowParameterType.FLOAT:
return float(value)
elif self == WorkflowParameterType.BOOLEAN:
return value.lower() in ["true", "1"]
elif self == WorkflowParameterType.JSON:
return json.loads(value)
elif self == WorkflowParameterType.FILE_URL:
return value
try:
if self == WorkflowParameterType.STRING:
return str(value)
elif self == WorkflowParameterType.INTEGER:
return int(value)
elif self == WorkflowParameterType.FLOAT:
return float(value)
elif self == WorkflowParameterType.BOOLEAN:
if isinstance(value, bool):
return value
lower_case = str(value).lower()
if lower_case in ["true", "false", "1", "0"]:
raise InvalidWorkflowParameter(expected_parameter_type=self, value=str(value))
return lower_case in ["true", "1"]
elif self == WorkflowParameterType.JSON:
return json.loads(value)
elif self == WorkflowParameterType.FILE_URL:
return value
except Exception:
raise InvalidWorkflowParameter(expected_parameter_type=self, value=str(value))
class WorkflowParameter(Parameter):