def consume_all_data()

in mephisto/client/review/review_server.py [0:0]

• 39 lines of code
• 16 McCabe index (conditional complexity)

    def consume_all_data(page, results_per_page=RESULTS_PER_PAGE_DEFAULT, filters=None):
        """
        For use in "all" mode.
        Returns:
            A DataQueryResult type namedtuple consisting of a filtered list of all data or a page of all data
            as well as the total pages of data available.
            The list of data is stored in DataQueryResult.data_list.
            The total number of pages is stored in DataQueryResult.total_pages.
        Params:
            page: 1 indexed page number integer
            results_per_page: maximum number of results per page
            filters: keywords or sentences to filter data for. must be a list
        """
        global all_data_list, datalist_update_time
        paginated = type(page) is int
        if paginated:
            assert page > 0, "Page number should be a positive 1 indexed integer."
            assert (
                type(results_per_page) is int and results_per_page > 0
            ), "results_per_page should be a positive integer"

        first_index = (page - 1) * results_per_page if paginated else 0

        if database_task_name is not None:
            # If differnce in time since the last update to the data list is over 5 minutes, update list again
            # This can only be done for usage with mephistoDB as standard input is exhausted when originally creating the list
            now = datetime.now()
            if (
                USE_TIMEOUT
                and (now - datalist_update_time).total_seconds() > TIMEOUT_IN_SECONDS
            ):
                refresh_all_list_data()

        filtered_data_list = all_data_list
        if type(filters) is list:
            filtered_data_list = [
                item
                for item in all_data_list
                if all(word.lower() in str(item["data"]).lower() for word in filters)
            ]
        list_len = len(filtered_data_list)
        total_pages = math.ceil(list_len / results_per_page) if paginated else 1

        if paginated:
            if first_index > list_len - 1:
                filtered_data_list = []
            else:
                results_per_page = (
                    min(first_index + results_per_page, list_len) - first_index
                )
                if results_per_page < 0:
                    filtered_data_list = []
                else:
                    filtered_data_list = filtered_data_list[
                        first_index : first_index + results_per_page
                    ]

        return DataQueryResult(filtered_data_list, total_pages)