Skip to content

update_flow_template

update_flow_template(
    id: Urn,
    definition: DefinitionDocument,
    *,
    config_overrides: Optional[IoTThingsGraphClientConfig] = None,
    compatible_namespace_version: Optional[Version] = None,
) -> UpdateFlowTemplateResponse

Updates the specified workflow. All deployed systems and system instances that use the workflow will see the changes in the flow when it is redeployed. If you don't want this behavior, copy the workflow (creating a new workflow with a different ID), and update the copy. The workflow can contain only entities in the specified namespace.

Parameters:

  • id (Urn) –

    The ID of the workflow to be updated.

    The ID should be in the following format.

    urn:tdm:REGION/ACCOUNT ID/default:workflow:WORKFLOWNAME

  • definition (DefinitionDocument) –

    The DefinitionDocument that contains the updated workflow definition.

  • compatible_namespace_version (Optional[Version], default: None ) –

    The version of the user's namespace.

    If no value is specified, the latest version is used by default. Use the GetFlowTemplateRevisions if you want to find earlier revisions of the flow to update.

Raises:

Source code in docs_src/capo_iotthingsgraph/_services/io_t_things_graph.py
def update_flow_template(
    self,
    id: "capo_iotthingsgraph.types.urn.Urn",
    definition: "capo_iotthingsgraph.types.definition_document.DefinitionDocument",
    *,
    config_overrides: Optional[IoTThingsGraphClientConfig] = None,
    compatible_namespace_version: Optional[
        "capo_iotthingsgraph.types.version.Version"
    ] = None,
) -> "capo_iotthingsgraph.types.update_flow_template_response.UpdateFlowTemplateResponse":
    """<p>Updates the specified workflow. All deployed systems and system instances that use the workflow will see the changes in the flow when it is redeployed. If you don't want this behavior, copy the workflow (creating a new workflow with a different ID), and update the copy. The workflow can contain only entities in the specified namespace. </p>

    Args:
        id: <p>The ID of the workflow to be updated.</p> <p>The ID should be in the following format.</p> <p> <code>urn:tdm:REGION/ACCOUNT ID/default:workflow:WORKFLOWNAME</code> </p>
        definition: <p>The <code>DefinitionDocument</code> that contains the updated workflow definition.</p>
        compatible_namespace_version: <p>The version of the user's namespace.</p> <p>If no value is specified, the latest version is used by default. Use the <code>GetFlowTemplateRevisions</code> if you want to find earlier revisions of the flow to update.</p>

    Raises:
        capo_iotthingsgraph.errors.internal_failure_exception.InternalFailureException: <p></p>
        capo_iotthingsgraph.errors.invalid_request_exception.InvalidRequestException: <p></p>
        capo_iotthingsgraph.errors.resource_not_found_exception.ResourceNotFoundException: <p></p>
        capo_iotthingsgraph.errors.throttling_exception.ThrottlingException: <p></p>
        capo_iotthingsgraph.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_iotthingsgraph.types.update_flow_template_request.UpdateFlowTemplateRequest]",
    ) -> OperationResponse[
        "capo_iotthingsgraph.types.update_flow_template_response.UpdateFlowTemplateResponse"
    ]:
        import capo_iotthingsgraph._operations.iot_things_graph_front_end_service.update_flow_template

        output, http_response = (
            capo_iotthingsgraph._operations.iot_things_graph_front_end_service.update_flow_template.update_flow_template(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("UpdateFlowTemplate")
    )
    input_: capo_iotthingsgraph.types.update_flow_template_request.UpdateFlowTemplateRequest = {
        "id": id,
        "definition": definition,
    }
    if compatible_namespace_version is not None:
        input_["compatible_namespace_version"] = compatible_namespace_version

    response = execute_pipeline(
        OperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    response.response.close()
    return response.output