create_workflow async ¶

async def create_workflow(
    self,
    steps: "capo_transfer.types.workflow_steps.WorkflowSteps",
    *,
    config_overrides: Optional[AsyncTransferClientConfig] = None,
    description: Optional[
        "capo_transfer.types.workflow_description.WorkflowDescription"
    ] = None,
    on_exception_steps: Optional[
        "capo_transfer.types.workflow_steps.WorkflowSteps"
    ] = None,
    tags: Optional["capo_transfer.types.tags.Tags"] = None,
    structured_log_destinations: Optional[
        "capo_transfer.types.structured_log_destinations.StructuredLogDestinations"
    ] = None,
) -> "capo_transfer.types.create_workflow_response.CreateWorkflowResponse":
    """<p> Allows you to create a workflow with specified steps and step details the workflow invokes after file transfer completes. After creating a workflow, you can associate the workflow created with any transfer servers by specifying the <code>workflow-details</code> field in <code>CreateServer</code> and <code>UpdateServer</code> operations. </p>

    Args:
        description: <p>A textual description for the workflow.</p>
        steps: <p>Specifies the details for the steps that are in the specified workflow.</p> <p> The <code>TYPE</code> specifies which of the following actions is being taken for this step. </p> <ul> <li> <p> <b> <code>COPY</code> </b> - Copy the file to another location.</p> </li> <li> <p> <b> <code>CUSTOM</code> </b> - Perform a custom step with an Lambda function target.</p> </li> <li> <p> <b> <code>DECRYPT</code> </b> - Decrypt a file that was encrypted before it was uploaded.</p> </li> <li> <p> <b> <code>DELETE</code> </b> - Delete the file.</p> </li> <li> <p> <b> <code>TAG</code> </b> - Add a tag to the file.</p> </li> </ul> <note> <p> Currently, copying and tagging are supported only on S3. </p> </note> <p> For file location, you specify either the Amazon S3 bucket and key, or the Amazon EFS file system ID and path. </p>
        on_exception_steps: <p>Specifies the steps (actions) to take if errors are encountered during execution of the workflow.</p> <note> <p>For custom steps, the Lambda function needs to send <code>FAILURE</code> to the call back API to kick off the exception steps. Additionally, if the Lambda does not send <code>SUCCESS</code> before it times out, the exception steps are executed.</p> </note>
        tags: <p>Key-value pairs that can be used to group and search for workflows. Tags are metadata attached to workflows for any purpose.</p>
        structured_log_destinations: <p>Specifies the log groups to which your workflow logs are sent.</p> <p>To specify a log group, you must provide the ARN for an existing log group. In this case, the format of the log group is as follows:</p> <p> <code>arn:partition:logs:region-name:amazon-account-id:log-group:log-group-name:*</code> </p> <p>For example, <code>arn:aws:logs:us-east-1:111122223333:log-group:mytestgroup:*</code> </p>

    Raises:
        capo_transfer.errors.access_denied_exception.AccessDeniedException: <p>You do not have sufficient access to perform this action.</p>
        capo_transfer.errors.internal_service_error.InternalServiceError: <p>This exception is thrown when an error occurs in the Transfer Family service.</p>
        capo_transfer.errors.invalid_request_exception.InvalidRequestException: <p>This exception is thrown when the client submits a malformed request.</p>
        capo_transfer.errors.resource_exists_exception.ResourceExistsException: <p>The requested resource does not exist, or exists in a region other than the one specified for the command.</p>
        capo_transfer.errors.service_unavailable_exception.ServiceUnavailableException: <p>The request has failed because the Amazon Web ServicesTransfer Family service is not available.</p>
        capo_transfer.errors.throttling_exception.ThrottlingException: <p>The request was denied due to request throttling.</p>
        capo_transfer.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_transfer.types.create_workflow_request.CreateWorkflowRequest]",
    ) -> AsyncOperationResponse[
        "capo_transfer.types.create_workflow_response.CreateWorkflowResponse"
    ]:
        import capo_transfer._operations.transfer_service.create_workflow

        (
            output,
            http_response,
        ) = await capo_transfer._operations.transfer_service.create_workflow.async_create_workflow(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("CreateWorkflow")
    )
    input_: capo_transfer.types.create_workflow_request.CreateWorkflowRequest = {
        "steps": steps
    }
    if description is not None:
        input_["description"] = description
    if on_exception_steps is not None:
        input_["on_exception_steps"] = on_exception_steps
    if tags is not None:
        input_["tags"] = tags
    if structured_log_destinations is not None:
        input_["structured_log_destinations"] = structured_log_destinations

    response = await aexecute_pipeline(
        AsyncOperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    await response.response.aclose()
    return response.output