"""Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT.""" # @generated-id: e2a0381191f6 from .basesdk import BaseSDK from .sdkconfiguration import SDKConfiguration from jsonpath import JSONPath from mistralai.client import errors, models, utils from mistralai.client._hooks import HookContext from mistralai.client.deployments import Deployments from mistralai.client.executions import Executions from mistralai.client.metrics import Metrics from mistralai.client.runs import Runs from mistralai.client.schedules import Schedules from mistralai.client.types import OptionalNullable, UNSET from mistralai.client.utils import get_security_from_env from mistralai.client.utils.unmarshal_json_response import unmarshal_json_response from mistralai.client.workflows_events import WorkflowsEvents from typing import Any, Awaitable, Dict, List, Mapping, Optional, Union from typing_extensions import deprecated # region imports import asyncio from pydantic import BaseModel import time # endregion imports class Workflows(BaseSDK): executions: Executions metrics: Metrics runs: Runs schedules: Schedules events: WorkflowsEvents deployments: Deployments def __init__( self, sdk_config: SDKConfiguration, parent_ref: Optional[object] = None ) -> None: BaseSDK.__init__(self, sdk_config, parent_ref=parent_ref) self.sdk_configuration = sdk_config self._init_sdks() def _init_sdks(self): self.executions = Executions(self.sdk_configuration, parent_ref=self.parent_ref) self.metrics = Metrics(self.sdk_configuration, parent_ref=self.parent_ref) self.runs = Runs(self.sdk_configuration, parent_ref=self.parent_ref) self.schedules = Schedules(self.sdk_configuration, parent_ref=self.parent_ref) self.events = WorkflowsEvents( self.sdk_configuration, parent_ref=self.parent_ref ) self.deployments = Deployments( self.sdk_configuration, parent_ref=self.parent_ref ) # region sdk-class-body def execute_workflow_and_wait( self, workflow_identifier: str, input: OptionalNullable[Dict[str, Any] | BaseModel] = UNSET, execution_id: OptionalNullable[str] = UNSET, deployment_name: OptionalNullable[str] = UNSET, custom_tracing_attributes: OptionalNullable[Dict[str, str]] = UNSET, task_queue: OptionalNullable[str] = UNSET, polling_interval: int = 5, max_attempts: Optional[int] = None, use_api_sync: bool = False, timeout_seconds: OptionalNullable[float] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> Any: """Execute a workflow and wait for its completion. Args: workflow_identifier: The workflow name or ID. input: The input to the workflow. This should be a dictionary or a BaseModel that matches the workflow's input schema. execution_id: Optional custom execution ID deployment_name: Name of the deployment to route this execution to custom_tracing_attributes: Custom tracing attributes task_queue: Deprecated. Use deployment_name instead. polling_interval: Seconds between status checks when polling max_attempts: Maximum number of polling attempts when polling (None for unlimited) use_api_sync: Whether to use the API's built-in sync execution capability timeout_seconds: Maximum time to wait in seconds when using API sync retries: Override the default retry configuration for this method server_url: Override the default server URL for this method timeout_ms: Override the default request timeout configuration for this method in milliseconds http_headers: Additional headers to set or replace on requests. Returns: The workflow result directly Raises: TimeoutError: If max_attempts is reached and workflow is still running RuntimeError: If workflow fails or terminates abnormally """ if use_api_sync: # Use the API's built-in synchronous execution response = self.execute_workflow( workflow_identifier=workflow_identifier, input=input, execution_id=execution_id, wait_for_result=True, timeout_seconds=timeout_seconds, custom_tracing_attributes=custom_tracing_attributes, task_queue=task_queue, deployment_name=deployment_name, retries=retries, server_url=server_url, timeout_ms=timeout_ms, http_headers=http_headers, ) return response.result # Use polling method execution = self.execute_workflow( workflow_identifier=workflow_identifier, input=input, execution_id=execution_id, custom_tracing_attributes=custom_tracing_attributes, task_queue=task_queue, deployment_name=deployment_name, retries=retries, server_url=server_url, timeout_ms=timeout_ms, http_headers=http_headers, ) # Wait for completion final_execution = self.wait_for_workflow_completion( execution.execution_id, polling_interval, max_attempts ) return final_execution.result def wait_for_workflow_completion( self, execution_id: str, polling_interval: int = 5, max_attempts: Optional[int] = None, ) -> models.WorkflowExecutionResponse: """Wait for a workflow to complete by polling its status. Args: execution_id: Execution ID of the workflow polling_interval: Seconds between status checks max_attempts: Maximum number of polling attempts (None for unlimited) Returns: WorkflowExecutionResponse with the final execution details Raises: TimeoutError: If max_attempts is reached and workflow is still running RuntimeError: If workflow fails or terminates abnormally """ attempts = 0 while True: response = self.executions.get_workflow_execution(execution_id=execution_id) if response.status != "RUNNING": if response.status == "COMPLETED": return response raise RuntimeError(f"Workflow failed with status: {response.status}") attempts += 1 if max_attempts is not None and attempts >= max_attempts: raise TimeoutError( f"Workflow is still running after {max_attempts} polling attempts" ) time.sleep(polling_interval) async def execute_workflow_and_wait_async( self, workflow_identifier: str, input: OptionalNullable[Dict[str, Any] | BaseModel] = UNSET, execution_id: OptionalNullable[str] = UNSET, deployment_name: OptionalNullable[str] = UNSET, custom_tracing_attributes: OptionalNullable[Dict[str, str]] = UNSET, task_queue: OptionalNullable[str] = UNSET, polling_interval: int = 5, max_attempts: Optional[int] = None, use_api_sync: bool = False, timeout_seconds: OptionalNullable[float] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> Any: """Execute a workflow and wait for its completion (async version). Args: workflow_identifier: The workflow name or ID. input: The input to the workflow. This should be a dictionary or a BaseModel that matches the workflow's input schema. execution_id: Optional custom execution ID deployment_name: Name of the deployment to route this execution to custom_tracing_attributes: Custom tracing attributes task_queue: Deprecated. Use deployment_name instead. polling_interval: Seconds between status checks when polling max_attempts: Maximum number of polling attempts when polling (None for unlimited) use_api_sync: Whether to use the API's built-in sync execution capability timeout_seconds: Maximum time to wait in seconds when using API sync retries: Override the default retry configuration for this method server_url: Override the default server URL for this method timeout_ms: Override the default request timeout configuration for this method in milliseconds http_headers: Additional headers to set or replace on requests. Returns: The workflow result directly Raises: TimeoutError: If max_attempts is reached and workflow is still running RuntimeError: If workflow fails or terminates abnormally """ if use_api_sync: # Use the API's built-in synchronous execution response = await self.execute_workflow_async( workflow_identifier=workflow_identifier, input=input, execution_id=execution_id, wait_for_result=True, timeout_seconds=timeout_seconds, custom_tracing_attributes=custom_tracing_attributes, task_queue=task_queue, deployment_name=deployment_name, retries=retries, server_url=server_url, timeout_ms=timeout_ms, http_headers=http_headers, ) return response.result # Use polling method execution = await self.execute_workflow_async( workflow_identifier=workflow_identifier, input=input, execution_id=execution_id, custom_tracing_attributes=custom_tracing_attributes, task_queue=task_queue, deployment_name=deployment_name, retries=retries, server_url=server_url, timeout_ms=timeout_ms, http_headers=http_headers, ) # Wait for completion final_execution = await self.wait_for_workflow_completion_async( execution.execution_id, polling_interval, max_attempts ) return final_execution.result async def wait_for_workflow_completion_async( self, execution_id: str, polling_interval: int = 5, max_attempts: Optional[int] = None, ) -> models.WorkflowExecutionResponse: """Wait for a workflow to complete by polling its status (async version). Args: execution_id: Execution ID of the workflow polling_interval: Seconds between status checks max_attempts: Maximum number of polling attempts (None for unlimited) Returns: WorkflowExecutionResponse with the final execution details Raises: TimeoutError: If max_attempts is reached and workflow is still running RuntimeError: If workflow fails or terminates abnormally """ attempts = 0 while True: response = await self.executions.get_workflow_execution_async( execution_id=execution_id ) if response.status != "RUNNING": if response.status == "COMPLETED": return response raise RuntimeError(f"Workflow failed with status: {response.status}") attempts += 1 if max_attempts is not None and attempts >= max_attempts: raise TimeoutError( f"Workflow is still running after {max_attempts} polling attempts" ) await asyncio.sleep(polling_interval) # endregion sdk-class-body def get_workflows( self, *, active_only: Optional[bool] = False, include_shared: Optional[bool] = True, available_in_chat_assistant: OptionalNullable[bool] = UNSET, archived: OptionalNullable[bool] = UNSET, cursor: OptionalNullable[str] = UNSET, limit: Optional[int] = 50, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> Optional[models.GetWorkflowsV1WorkflowsGetResponse]: r"""Get Workflows :param active_only: Whether to only return active workflows :param include_shared: Whether to include shared workflows :param available_in_chat_assistant: Whether to only return workflows available in chat assistant :param archived: Filter by archived state. False=exclude archived, True=only archived, None=include all :param cursor: The cursor for pagination :param limit: The maximum number of workflows to return :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetWorkflowsV1WorkflowsGetRequest( active_only=active_only, include_shared=include_shared, available_in_chat_assistant=available_in_chat_assistant, archived=archived, cursor=cursor, limit=limit, ) req = self._build_request( method="GET", path="/v1/workflows", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=False, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="get_workflows_v1_workflows_get", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) def next_func() -> Optional[models.GetWorkflowsV1WorkflowsGetResponse]: body = utils.unmarshal_json(http_res.text, Union[Dict[Any, Any], List[Any]]) next_cursor = JSONPath("$.next_cursor").parse(body) if len(next_cursor) == 0: return None next_cursor = next_cursor[0] if next_cursor is None or str(next_cursor).strip() == "": return None results = JSONPath("$.workflows").parse(body) if len(results) == 0 or len(results[0]) == 0: return None limit = request.limit if isinstance(request.limit, int) else 50 if len(results[0]) < limit: return None return self.get_workflows( active_only=active_only, include_shared=include_shared, available_in_chat_assistant=available_in_chat_assistant, archived=archived, cursor=next_cursor, limit=limit, retries=retries, server_url=server_url, timeout_ms=timeout_ms, http_headers=http_headers, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return models.GetWorkflowsV1WorkflowsGetResponse( result=unmarshal_json_response(models.WorkflowListResponse, http_res), next=next_func, ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) async def get_workflows_async( self, *, active_only: Optional[bool] = False, include_shared: Optional[bool] = True, available_in_chat_assistant: OptionalNullable[bool] = UNSET, archived: OptionalNullable[bool] = UNSET, cursor: OptionalNullable[str] = UNSET, limit: Optional[int] = 50, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> Optional[models.GetWorkflowsV1WorkflowsGetResponse]: r"""Get Workflows :param active_only: Whether to only return active workflows :param include_shared: Whether to include shared workflows :param available_in_chat_assistant: Whether to only return workflows available in chat assistant :param archived: Filter by archived state. False=exclude archived, True=only archived, None=include all :param cursor: The cursor for pagination :param limit: The maximum number of workflows to return :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetWorkflowsV1WorkflowsGetRequest( active_only=active_only, include_shared=include_shared, available_in_chat_assistant=available_in_chat_assistant, archived=archived, cursor=cursor, limit=limit, ) req = self._build_request_async( method="GET", path="/v1/workflows", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=False, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="get_workflows_v1_workflows_get", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) def next_func() -> ( Awaitable[Optional[models.GetWorkflowsV1WorkflowsGetResponse]] ): body = utils.unmarshal_json(http_res.text, Union[Dict[Any, Any], List[Any]]) async def empty_result(): return None next_cursor = JSONPath("$.next_cursor").parse(body) if len(next_cursor) == 0: return empty_result() next_cursor = next_cursor[0] if next_cursor is None or str(next_cursor).strip() == "": return empty_result() results = JSONPath("$.workflows").parse(body) if len(results) == 0 or len(results[0]) == 0: return empty_result() limit = request.limit if isinstance(request.limit, int) else 50 if len(results[0]) < limit: return empty_result() return self.get_workflows_async( active_only=active_only, include_shared=include_shared, available_in_chat_assistant=available_in_chat_assistant, archived=archived, cursor=next_cursor, limit=limit, retries=retries, server_url=server_url, timeout_ms=timeout_ms, http_headers=http_headers, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return models.GetWorkflowsV1WorkflowsGetResponse( result=unmarshal_json_response(models.WorkflowListResponse, http_res), next=next_func, ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) def get_workflow_registrations( self, *, workflow_id: OptionalNullable[str] = UNSET, task_queue: OptionalNullable[str] = UNSET, active_only: Optional[bool] = False, include_shared: Optional[bool] = True, workflow_search: OptionalNullable[str] = UNSET, archived: OptionalNullable[bool] = UNSET, with_workflow: Optional[bool] = False, available_in_chat_assistant: OptionalNullable[bool] = UNSET, limit: Optional[int] = 50, cursor: OptionalNullable[str] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowRegistrationListResponse: r"""Get Workflow Registrations :param workflow_id: The workflow ID to filter by :param task_queue: The task queue to filter by :param active_only: Whether to only return active workflows versions :param include_shared: Whether to include shared workflow versions :param workflow_search: The workflow name to filter by :param archived: Filter by archived state. False=exclude archived, True=only archived, None=include all :param with_workflow: Whether to include the workflow definition :param available_in_chat_assistant: Whether to only return workflows available in chat assistant :param limit: The maximum number of workflows versions to return :param cursor: The cursor for pagination :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetWorkflowRegistrationsV1WorkflowsRegistrationsGetRequest( workflow_id=workflow_id, task_queue=task_queue, active_only=active_only, include_shared=include_shared, workflow_search=workflow_search, archived=archived, with_workflow=with_workflow, available_in_chat_assistant=available_in_chat_assistant, limit=limit, cursor=cursor, ) req = self._build_request( method="GET", path="/v1/workflows/registrations", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=False, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="get_workflow_registrations_v1_workflows_registrations_get", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( models.WorkflowRegistrationListResponse, http_res ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) async def get_workflow_registrations_async( self, *, workflow_id: OptionalNullable[str] = UNSET, task_queue: OptionalNullable[str] = UNSET, active_only: Optional[bool] = False, include_shared: Optional[bool] = True, workflow_search: OptionalNullable[str] = UNSET, archived: OptionalNullable[bool] = UNSET, with_workflow: Optional[bool] = False, available_in_chat_assistant: OptionalNullable[bool] = UNSET, limit: Optional[int] = 50, cursor: OptionalNullable[str] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowRegistrationListResponse: r"""Get Workflow Registrations :param workflow_id: The workflow ID to filter by :param task_queue: The task queue to filter by :param active_only: Whether to only return active workflows versions :param include_shared: Whether to include shared workflow versions :param workflow_search: The workflow name to filter by :param archived: Filter by archived state. False=exclude archived, True=only archived, None=include all :param with_workflow: Whether to include the workflow definition :param available_in_chat_assistant: Whether to only return workflows available in chat assistant :param limit: The maximum number of workflows versions to return :param cursor: The cursor for pagination :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetWorkflowRegistrationsV1WorkflowsRegistrationsGetRequest( workflow_id=workflow_id, task_queue=task_queue, active_only=active_only, include_shared=include_shared, workflow_search=workflow_search, archived=archived, with_workflow=with_workflow, available_in_chat_assistant=available_in_chat_assistant, limit=limit, cursor=cursor, ) req = self._build_request_async( method="GET", path="/v1/workflows/registrations", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=False, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="get_workflow_registrations_v1_workflows_registrations_get", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( models.WorkflowRegistrationListResponse, http_res ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) def execute_workflow( self, *, workflow_identifier: str, execution_id: OptionalNullable[str] = UNSET, input: OptionalNullable[Any] = UNSET, encoded_input: OptionalNullable[ Union[models.NetworkEncodedInput, models.NetworkEncodedInputTypedDict] ] = UNSET, wait_for_result: Optional[bool] = False, timeout_seconds: OptionalNullable[float] = UNSET, custom_tracing_attributes: OptionalNullable[Dict[str, str]] = UNSET, extensions: OptionalNullable[Dict[str, Any]] = UNSET, task_queue: OptionalNullable[str] = UNSET, deployment_name: OptionalNullable[str] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.ResponseExecuteWorkflowV1WorkflowsWorkflowIdentifierExecutePost: r"""Execute Workflow :param workflow_identifier: :param execution_id: Allows you to specify a custom execution ID. If not provided, a random ID will be generated. :param input: The input to the workflow. This should be a dictionary or a BaseModel that matches the workflow's input schema. :param encoded_input: Encoded input to the workflow, used when payload encoding is enabled. :param wait_for_result: If true, wait for the workflow to complete and return the result directly. :param timeout_seconds: Maximum time to wait for completion when wait_for_result is true. :param custom_tracing_attributes: :param extensions: Plugin-specific data to propagate into WorkflowContext.extensions at execution time. :param task_queue: Deprecated. Use deployment_name instead. :param deployment_name: Name of the deployment to route this execution to :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.ExecuteWorkflowV1WorkflowsWorkflowIdentifierExecutePostRequest( workflow_identifier=workflow_identifier, workflow_execution_request=models.WorkflowExecutionRequest( execution_id=execution_id, input=input, encoded_input=utils.get_pydantic_model( encoded_input, OptionalNullable[models.NetworkEncodedInput] ), wait_for_result=wait_for_result, timeout_seconds=timeout_seconds, custom_tracing_attributes=custom_tracing_attributes, extensions=extensions, task_queue=task_queue, deployment_name=deployment_name, ), ) req = self._build_request( method="POST", path="/v1/workflows/{workflow_identifier}/execute", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.workflow_execution_request, False, False, "json", models.WorkflowExecutionRequest, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="execute_workflow_v1_workflows__workflow_identifier__execute_post", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( models.ResponseExecuteWorkflowV1WorkflowsWorkflowIdentifierExecutePost, http_res, ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) async def execute_workflow_async( self, *, workflow_identifier: str, execution_id: OptionalNullable[str] = UNSET, input: OptionalNullable[Any] = UNSET, encoded_input: OptionalNullable[ Union[models.NetworkEncodedInput, models.NetworkEncodedInputTypedDict] ] = UNSET, wait_for_result: Optional[bool] = False, timeout_seconds: OptionalNullable[float] = UNSET, custom_tracing_attributes: OptionalNullable[Dict[str, str]] = UNSET, extensions: OptionalNullable[Dict[str, Any]] = UNSET, task_queue: OptionalNullable[str] = UNSET, deployment_name: OptionalNullable[str] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.ResponseExecuteWorkflowV1WorkflowsWorkflowIdentifierExecutePost: r"""Execute Workflow :param workflow_identifier: :param execution_id: Allows you to specify a custom execution ID. If not provided, a random ID will be generated. :param input: The input to the workflow. This should be a dictionary or a BaseModel that matches the workflow's input schema. :param encoded_input: Encoded input to the workflow, used when payload encoding is enabled. :param wait_for_result: If true, wait for the workflow to complete and return the result directly. :param timeout_seconds: Maximum time to wait for completion when wait_for_result is true. :param custom_tracing_attributes: :param extensions: Plugin-specific data to propagate into WorkflowContext.extensions at execution time. :param task_queue: Deprecated. Use deployment_name instead. :param deployment_name: Name of the deployment to route this execution to :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.ExecuteWorkflowV1WorkflowsWorkflowIdentifierExecutePostRequest( workflow_identifier=workflow_identifier, workflow_execution_request=models.WorkflowExecutionRequest( execution_id=execution_id, input=input, encoded_input=utils.get_pydantic_model( encoded_input, OptionalNullable[models.NetworkEncodedInput] ), wait_for_result=wait_for_result, timeout_seconds=timeout_seconds, custom_tracing_attributes=custom_tracing_attributes, extensions=extensions, task_queue=task_queue, deployment_name=deployment_name, ), ) req = self._build_request_async( method="POST", path="/v1/workflows/{workflow_identifier}/execute", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.workflow_execution_request, False, False, "json", models.WorkflowExecutionRequest, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="execute_workflow_v1_workflows__workflow_identifier__execute_post", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( models.ResponseExecuteWorkflowV1WorkflowsWorkflowIdentifierExecutePost, http_res, ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) @deprecated( "warning: ** DEPRECATED ** - This will be removed in a future release, please migrate away from it as soon as possible." ) def execute_workflow_registration( self, *, workflow_registration_id: str, execution_id: OptionalNullable[str] = UNSET, input: OptionalNullable[Any] = UNSET, encoded_input: OptionalNullable[ Union[models.NetworkEncodedInput, models.NetworkEncodedInputTypedDict] ] = UNSET, wait_for_result: Optional[bool] = False, timeout_seconds: OptionalNullable[float] = UNSET, custom_tracing_attributes: OptionalNullable[Dict[str, str]] = UNSET, extensions: OptionalNullable[Dict[str, Any]] = UNSET, task_queue: OptionalNullable[str] = UNSET, deployment_name: OptionalNullable[str] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.ResponseExecuteWorkflowRegistrationV1WorkflowsRegistrationsWorkflowRegistrationIDExecutePost: r"""Execute Workflow Registration :param workflow_registration_id: :param execution_id: Allows you to specify a custom execution ID. If not provided, a random ID will be generated. :param input: The input to the workflow. This should be a dictionary or a BaseModel that matches the workflow's input schema. :param encoded_input: Encoded input to the workflow, used when payload encoding is enabled. :param wait_for_result: If true, wait for the workflow to complete and return the result directly. :param timeout_seconds: Maximum time to wait for completion when wait_for_result is true. :param custom_tracing_attributes: :param extensions: Plugin-specific data to propagate into WorkflowContext.extensions at execution time. :param task_queue: Deprecated. Use deployment_name instead. :param deployment_name: Name of the deployment to route this execution to :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.ExecuteWorkflowRegistrationV1WorkflowsRegistrationsWorkflowRegistrationIDExecutePostRequest( workflow_registration_id=workflow_registration_id, workflow_execution_request=models.WorkflowExecutionRequest( execution_id=execution_id, input=input, encoded_input=utils.get_pydantic_model( encoded_input, OptionalNullable[models.NetworkEncodedInput] ), wait_for_result=wait_for_result, timeout_seconds=timeout_seconds, custom_tracing_attributes=custom_tracing_attributes, extensions=extensions, task_queue=task_queue, deployment_name=deployment_name, ), ) req = self._build_request( method="POST", path="/v1/workflows/registrations/{workflow_registration_id}/execute", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.workflow_execution_request, False, False, "json", models.WorkflowExecutionRequest, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="execute_workflow_registration_v1_workflows_registrations__workflow_registration_id__execute_post", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( models.ResponseExecuteWorkflowRegistrationV1WorkflowsRegistrationsWorkflowRegistrationIDExecutePost, http_res, ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) @deprecated( "warning: ** DEPRECATED ** - This will be removed in a future release, please migrate away from it as soon as possible." ) async def execute_workflow_registration_async( self, *, workflow_registration_id: str, execution_id: OptionalNullable[str] = UNSET, input: OptionalNullable[Any] = UNSET, encoded_input: OptionalNullable[ Union[models.NetworkEncodedInput, models.NetworkEncodedInputTypedDict] ] = UNSET, wait_for_result: Optional[bool] = False, timeout_seconds: OptionalNullable[float] = UNSET, custom_tracing_attributes: OptionalNullable[Dict[str, str]] = UNSET, extensions: OptionalNullable[Dict[str, Any]] = UNSET, task_queue: OptionalNullable[str] = UNSET, deployment_name: OptionalNullable[str] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.ResponseExecuteWorkflowRegistrationV1WorkflowsRegistrationsWorkflowRegistrationIDExecutePost: r"""Execute Workflow Registration :param workflow_registration_id: :param execution_id: Allows you to specify a custom execution ID. If not provided, a random ID will be generated. :param input: The input to the workflow. This should be a dictionary or a BaseModel that matches the workflow's input schema. :param encoded_input: Encoded input to the workflow, used when payload encoding is enabled. :param wait_for_result: If true, wait for the workflow to complete and return the result directly. :param timeout_seconds: Maximum time to wait for completion when wait_for_result is true. :param custom_tracing_attributes: :param extensions: Plugin-specific data to propagate into WorkflowContext.extensions at execution time. :param task_queue: Deprecated. Use deployment_name instead. :param deployment_name: Name of the deployment to route this execution to :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.ExecuteWorkflowRegistrationV1WorkflowsRegistrationsWorkflowRegistrationIDExecutePostRequest( workflow_registration_id=workflow_registration_id, workflow_execution_request=models.WorkflowExecutionRequest( execution_id=execution_id, input=input, encoded_input=utils.get_pydantic_model( encoded_input, OptionalNullable[models.NetworkEncodedInput] ), wait_for_result=wait_for_result, timeout_seconds=timeout_seconds, custom_tracing_attributes=custom_tracing_attributes, extensions=extensions, task_queue=task_queue, deployment_name=deployment_name, ), ) req = self._build_request_async( method="POST", path="/v1/workflows/registrations/{workflow_registration_id}/execute", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.workflow_execution_request, False, False, "json", models.WorkflowExecutionRequest, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="execute_workflow_registration_v1_workflows_registrations__workflow_registration_id__execute_post", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( models.ResponseExecuteWorkflowRegistrationV1WorkflowsRegistrationsWorkflowRegistrationIDExecutePost, http_res, ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) def get_workflow( self, *, workflow_identifier: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowGetResponse: r"""Get Workflow :param workflow_identifier: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetWorkflowV1WorkflowsWorkflowIdentifierGetRequest( workflow_identifier=workflow_identifier, ) req = self._build_request( method="GET", path="/v1/workflows/{workflow_identifier}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="get_workflow_v1_workflows__workflow_identifier__get", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.WorkflowGetResponse, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) async def get_workflow_async( self, *, workflow_identifier: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowGetResponse: r"""Get Workflow :param workflow_identifier: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetWorkflowV1WorkflowsWorkflowIdentifierGetRequest( workflow_identifier=workflow_identifier, ) req = self._build_request_async( method="GET", path="/v1/workflows/{workflow_identifier}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="get_workflow_v1_workflows__workflow_identifier__get", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.WorkflowGetResponse, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) def update_workflow( self, *, workflow_identifier: str, display_name: OptionalNullable[str] = UNSET, description: OptionalNullable[str] = UNSET, available_in_chat_assistant: OptionalNullable[bool] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowUpdateResponse: r"""Update Workflow :param workflow_identifier: :param display_name: New display name value :param description: New description value :param available_in_chat_assistant: Whether to make the workflow available in the chat assistant :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.UpdateWorkflowV1WorkflowsWorkflowIdentifierPutRequest( workflow_identifier=workflow_identifier, workflow_update_request=models.WorkflowUpdateRequest( display_name=display_name, description=description, available_in_chat_assistant=available_in_chat_assistant, ), ) req = self._build_request( method="PUT", path="/v1/workflows/{workflow_identifier}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.workflow_update_request, False, False, "json", models.WorkflowUpdateRequest, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="update_workflow_v1_workflows__workflow_identifier__put", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.WorkflowUpdateResponse, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) async def update_workflow_async( self, *, workflow_identifier: str, display_name: OptionalNullable[str] = UNSET, description: OptionalNullable[str] = UNSET, available_in_chat_assistant: OptionalNullable[bool] = UNSET, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowUpdateResponse: r"""Update Workflow :param workflow_identifier: :param display_name: New display name value :param description: New description value :param available_in_chat_assistant: Whether to make the workflow available in the chat assistant :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.UpdateWorkflowV1WorkflowsWorkflowIdentifierPutRequest( workflow_identifier=workflow_identifier, workflow_update_request=models.WorkflowUpdateRequest( display_name=display_name, description=description, available_in_chat_assistant=available_in_chat_assistant, ), ) req = self._build_request_async( method="PUT", path="/v1/workflows/{workflow_identifier}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=True, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, get_serialized_body=lambda: utils.serialize_request_body( request.workflow_update_request, False, False, "json", models.WorkflowUpdateRequest, ), allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="update_workflow_v1_workflows__workflow_identifier__put", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.WorkflowUpdateResponse, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) def get_workflow_registration( self, *, workflow_registration_id: str, with_workflow: Optional[bool] = False, include_shared: Optional[bool] = True, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowRegistrationGetResponse: r"""Get Workflow Registration :param workflow_registration_id: :param with_workflow: Whether to include the workflow definition :param include_shared: Whether to include shared workflow versions :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetWorkflowRegistrationV1WorkflowsRegistrationsWorkflowRegistrationIDGetRequest( workflow_registration_id=workflow_registration_id, with_workflow=with_workflow, include_shared=include_shared, ) req = self._build_request( method="GET", path="/v1/workflows/registrations/{workflow_registration_id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="get_workflow_registration_v1_workflows_registrations__workflow_registration_id__get", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( models.WorkflowRegistrationGetResponse, http_res ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) async def get_workflow_registration_async( self, *, workflow_registration_id: str, with_workflow: Optional[bool] = False, include_shared: Optional[bool] = True, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowRegistrationGetResponse: r"""Get Workflow Registration :param workflow_registration_id: :param with_workflow: Whether to include the workflow definition :param include_shared: Whether to include shared workflow versions :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.GetWorkflowRegistrationV1WorkflowsRegistrationsWorkflowRegistrationIDGetRequest( workflow_registration_id=workflow_registration_id, with_workflow=with_workflow, include_shared=include_shared, ) req = self._build_request_async( method="GET", path="/v1/workflows/registrations/{workflow_registration_id}", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="get_workflow_registration_v1_workflows_registrations__workflow_registration_id__get", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response( models.WorkflowRegistrationGetResponse, http_res ) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) def archive_workflow( self, *, workflow_identifier: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowArchiveResponse: r"""Archive Workflow :param workflow_identifier: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.ArchiveWorkflowV1WorkflowsWorkflowIdentifierArchivePutRequest( workflow_identifier=workflow_identifier, ) req = self._build_request( method="PUT", path="/v1/workflows/{workflow_identifier}/archive", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="archive_workflow_v1_workflows__workflow_identifier__archive_put", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.WorkflowArchiveResponse, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) async def archive_workflow_async( self, *, workflow_identifier: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowArchiveResponse: r"""Archive Workflow :param workflow_identifier: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = models.ArchiveWorkflowV1WorkflowsWorkflowIdentifierArchivePutRequest( workflow_identifier=workflow_identifier, ) req = self._build_request_async( method="PUT", path="/v1/workflows/{workflow_identifier}/archive", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="archive_workflow_v1_workflows__workflow_identifier__archive_put", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.WorkflowArchiveResponse, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) def unarchive_workflow( self, *, workflow_identifier: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowUnarchiveResponse: r"""Unarchive Workflow :param workflow_identifier: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = ( models.UnarchiveWorkflowV1WorkflowsWorkflowIdentifierUnarchivePutRequest( workflow_identifier=workflow_identifier, ) ) req = self._build_request( method="PUT", path="/v1/workflows/{workflow_identifier}/unarchive", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = self.do_request( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="unarchive_workflow_v1_workflows__workflow_identifier__unarchive_put", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.WorkflowUnarchiveResponse, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = utils.stream_to_text(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res) async def unarchive_workflow_async( self, *, workflow_identifier: str, retries: OptionalNullable[utils.RetryConfig] = UNSET, server_url: Optional[str] = None, timeout_ms: Optional[int] = None, http_headers: Optional[Mapping[str, str]] = None, ) -> models.WorkflowUnarchiveResponse: r"""Unarchive Workflow :param workflow_identifier: :param retries: Override the default retry configuration for this method :param server_url: Override the default server URL for this method :param timeout_ms: Override the default request timeout configuration for this method in milliseconds :param http_headers: Additional headers to set or replace on requests. """ base_url = None url_variables = None if timeout_ms is None: timeout_ms = self.sdk_configuration.timeout_ms if timeout_ms is None: timeout_ms = 30000 if server_url is not None: base_url = server_url else: base_url = self._get_url(base_url, url_variables) request = ( models.UnarchiveWorkflowV1WorkflowsWorkflowIdentifierUnarchivePutRequest( workflow_identifier=workflow_identifier, ) ) req = self._build_request_async( method="PUT", path="/v1/workflows/{workflow_identifier}/unarchive", base_url=base_url, url_variables=url_variables, request=request, request_body_required=False, request_has_path_params=True, request_has_query_params=True, user_agent_header="user-agent", accept_header_value="application/json", http_headers=http_headers, security=self.sdk_configuration.security, allow_empty_value=None, timeout_ms=timeout_ms, ) if retries == UNSET: if self.sdk_configuration.retry_config is not UNSET: retries = self.sdk_configuration.retry_config retry_config = None if isinstance(retries, utils.RetryConfig): retry_config = (retries, ["429", "500", "502", "503", "504"]) http_res = await self.do_request_async( hook_ctx=HookContext( config=self.sdk_configuration, base_url=base_url or "", operation_id="unarchive_workflow_v1_workflows__workflow_identifier__unarchive_put", oauth2_scopes=None, security_source=get_security_from_env( self.sdk_configuration.security, models.Security ), ), request=req, error_status_codes=["422", "4XX", "5XX"], retry_config=retry_config, ) response_data: Any = None if utils.match_response(http_res, "200", "application/json"): return unmarshal_json_response(models.WorkflowUnarchiveResponse, http_res) if utils.match_response(http_res, "422", "application/json"): response_data = unmarshal_json_response( errors.HTTPValidationErrorData, http_res ) raise errors.HTTPValidationError(response_data, http_res) if utils.match_response(http_res, "4XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) if utils.match_response(http_res, "5XX", "*"): http_res_text = await utils.stream_to_text_async(http_res) raise errors.SDKError("API error occurred", http_res, http_res_text) raise errors.SDKError("Unexpected response received", http_res)