def list()

in treeherder/webapp/api/jobs.py [0:0]

• 45 lines of code
• 10 McCabe index (conditional complexity)

    def list(self, request, project):
        """
        GET method implementation for list view
        Optional parameters (default):
        - offset (0)
        - count (10)
        - return_type (dict)
        """
        max_jobs_count = 2000

        filter_params = {}

        # various hacks to ensure API backwards compatibility
        for param_key, param_value in request.query_params.items():
            # replace `result_set_id` with `push_id`
            if param_key.startswith("result_set_id"):
                new_param_key = param_key.replace("result_set_id", "push_id")
                filter_params[new_param_key] = param_value
            # convert legacy timestamp parameters to time ones
            elif param_key in ["submit_timestamp", "start_timestamp", "end_timestamp"]:
                new_param_key = param_key.replace("timestamp", "time")
                filter_params[new_param_key] = datetime.datetime.fromtimestamp(float(param_value))
            # sanity check 'last modified'
            elif param_key.startswith("last_modified"):
                try:
                    parser.parse(param_value)
                except ValueError:
                    return Response(
                        f"Invalid date value for `last_modified`: {param_value}",
                        status=HTTP_400_BAD_REQUEST,
                    )
                filter_params[param_key] = param_value
            # default case
            else:
                filter_params[param_key] = param_value

        try:
            offset = int(filter_params.get("offset", 0))
            count = int(filter_params.get("count", 10))
        except ValueError:
            return Response("Invalid value for offset or count", status=HTTP_400_BAD_REQUEST)
        return_type = filter_params.get("return_type", "dict").lower()

        if count > max_jobs_count:
            msg = f"Specified count exceeds API MAX_JOBS_COUNT value: {max_jobs_count}"
            return Response({"detail": msg}, status=HTTP_400_BAD_REQUEST)

        try:
            repository = Repository.objects.get(name=project)
        except Repository.DoesNotExist:
            return Response(
                {"detail": f"No project with name {project}"}, status=HTTP_404_NOT_FOUND
            )
        jobs = JobFilter(
            {k: v for (k, v) in filter_params.items()},
            queryset=Job.objects.filter(repository=repository).select_related(
                *self._default_select_related
            ),
        ).qs

        response_body = self._get_job_list_response(jobs, offset, count, return_type)
        response_body["meta"] = dict(repository=project, offset=offset, count=count)

        return Response(response_body)