get_activity_task ¶

def get_activity_task(
    self,
    activity_arn: "capo_sfn.types.arn.Arn",
    *,
    config_overrides: Optional[SFNClientConfig] = None,
    worker_name: Optional["capo_sfn.types.name.Name"] = None,
) -> "capo_sfn.types.get_activity_task_output.GetActivityTaskOutput":
    """<p>Used by workers to retrieve a task (with the specified activity ARN) which has been scheduled for execution by a running state machine. This initiates a long poll, where the service holds the HTTP connection open and responds as soon as a task becomes available (i.e. an execution of a task of this type is needed.) The maximum time the service holds on to the request before responding is 60 seconds. If no task is available within 60 seconds, the poll returns a <code>taskToken</code> with a null string.</p> <important> <p>Workers should set their client side socket timeout to at least 65 seconds (5 seconds higher than the maximum time the service may hold the poll request).</p> <p>Polling with <code>GetActivityTask</code> can cause latency in some implementations. See <a href="https://docs.aws.amazon.com/step-functions/latest/dg/bp-activity-pollers.html">Avoid Latency When Polling for Activity Tasks</a> in the Step Functions Developer Guide.</p> </important>

    Args:
        activity_arn: <p>The Amazon Resource Name (ARN) of the activity to retrieve tasks from (assigned when you create the task using <a>CreateActivity</a>.)</p>
        worker_name: <p>You can provide an arbitrary name in order to identify the worker that the task is assigned to. This name is used when it is logged in the execution history.</p>

    Raises:
        capo_sfn.errors.activity_does_not_exist.ActivityDoesNotExist: <p>The specified activity does not exist.</p>
        capo_sfn.errors.activity_worker_limit_exceeded.ActivityWorkerLimitExceeded: <p>The maximum number of workers concurrently polling for activity tasks has been reached.</p>
        capo_sfn.errors.invalid_arn.InvalidArn: <p>The provided Amazon Resource Name (ARN) is not valid.</p>
        capo_sfn.errors.kms_access_denied_exception.KmsAccessDeniedException: <p>Either your KMS key policy or API caller does not have the required permissions.</p>
        capo_sfn.errors.kms_invalid_state_exception.KmsInvalidStateException: <p>The KMS key is not in valid state, for example: Disabled or Deleted.</p>
        capo_sfn.errors.kms_throttling_exception.KmsThrottlingException: <p>Received when KMS returns <code>ThrottlingException</code> for a KMS call that Step Functions makes on behalf of the caller.</p>
        capo_sfn.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_sfn.types.get_activity_task_input.GetActivityTaskInput]",
    ) -> OperationResponse[
        "capo_sfn.types.get_activity_task_output.GetActivityTaskOutput"
    ]:
        import capo_sfn._operations.aws_step_functions.get_activity_task

        output, http_response = (
            capo_sfn._operations.aws_step_functions.get_activity_task.get_activity_task(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    def _redact_output(
        output: "capo_sfn.types.get_activity_task_output.GetActivityTaskOutput",
    ) -> dict[str, Any]:
        import capo_sfn.types.get_activity_task_output

        return capo_sfn.types.get_activity_task_output.redact_sensitive(output)

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging("GetActivityTask", redact_output=_redact_output),
    )
    input_: capo_sfn.types.get_activity_task_input.GetActivityTaskInput = {
        "activity_arn": activity_arn
    }
    if worker_name is not None:
        input_["worker_name"] = worker_name

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