def union_results()

in services/ui_backend_service/api/search.py [0:0]

• 7 lines of code
• 3 McCabe index (conditional complexity)

    def union_results(self, new_results: dict, request_identifier: str):
        """
        Union new results with existing results, and update the keyset to prevent duplicate results.
        Key is determined by step_name and task_id.
        """
        for result in new_results:
            key = result['step_name'] + '/' + result['task_id']
            if key in self.search_result_keyset:
                continue

            self.search_result_keyset[request_identifier].add(key)
            self.search_results[request_identifier].append(result)