def query_status()

in scheduler/job_requester/requester.py [0:0]

• 38 lines of code
• 12 McCabe index (conditional complexity)

    def query_status(self, identifier):
        """
        :param identifier: <Message object> unique identifier returned from call to send_request
        :return: <dict> {"status": <string> queuing/preparing/completed/failed/runtimeError,
                         "reason" (if status == failed): <string> maxRetries/timeout/duplicatePR,
                         "queueNum" (if status == queuing): <int>
                         }
        """
        retries = 2
        request_ticket_name = identifier.ticket_name
        ticket_without_extension = request_ticket_name.rstrip(".json")
        instance_type = identifier.instance_type
        job_type = identifier.job_type

        for _ in range(retries):
            # check if ticket is on the queue
            ticket_objects = self.s3_client.list_objects(
                Bucket=self.s3_ticket_bucket, Prefix="request_tickets/"
            )
            # "Contents" in the API response only if there are objects satisfy the prefix
            if "Contents" in ticket_objects:
                ticket_name_pattern = re.compile(".*\/(.*)")
                ticket_names_list = [
                    ticket_name_pattern.match(ticket["Key"]).group(1)
                    for ticket in ticket_objects["Contents"]
                    if ticket["Key"].endswith(".json")
                ]
                # ticket is on the queue, find the queue number
                if request_ticket_name in ticket_names_list:
                    ticket_names_list.sort(key=cmp_to_key(self.ticket_timestamp_cmp_function))
                    queue_num = ticket_names_list.index(request_ticket_name)
                    return self.construct_query_response("queuing", queueNum=queue_num)

            # check if ticket is on the dead letter queue
            ticket_in_dead_letter = self.search_ticket_folder(
                "dead_letter_queue", ticket_without_extension
            )
            if ticket_in_dead_letter:
                return ticket_in_dead_letter

            ticket_in_duplicate = self.search_ticket_folder(
                "duplicate_pr_requests", ticket_without_extension
            )
            if ticket_in_duplicate:
                return ticket_in_duplicate

            ticket_in_progress = self.search_ticket_folder(
                "resource_pool", f"{instance_type}-{job_type}/{ticket_without_extension}"
            )
            if ticket_in_progress:
                return ticket_in_progress

            time.sleep(2)

        raise AssertionError(f"Request ticket name {request_ticket_name} could not be found.")