nodes: 0, module; 1, function_definition; 2, function_name:delete_orphan_unique_identities; 3, parameters; 4, block; 5, identifier:es; 6, identifier:sortinghat_db; 7, identifier:current_data_source; 8, identifier:active_data_sources; 9, expression_statement; 10, function_definition; 11, function_definition; 12, function_definition; 13, function_definition; 14, expression_statement; 15, expression_statement; 16, expression_statement; 17, comment:# Collect all unique identities; 18, for_statement; 19, comment:# Check that no uuids have been left to process; 20, if_statement; 21, expression_statement; 22, expression_statement; 23, comment:"""Delete all unique identities which appear in SortingHat, but not in the IDENTITIES_INDEX.
  
      :param es: ElasticSearchDSL object
      :param sortinghat_db: instance of the SortingHat database
      :param current_data_source: current data source
      :param active_data_sources: list of active data sources
      """; 24, function_name:get_uuids_in_index; 25, parameters; 26, block; 27, function_name:delete_unique_identities; 28, parameters; 29, block; 30, function_name:delete_identities; 31, parameters; 32, block; 33, function_name:has_identities_in_data_sources; 34, parameters; 35, block; 36, assignment; 37, assignment; 38, assignment; 39, identifier:unique_identity; 40, call; 41, comment:# Remove a unique identity if all its identities are in non active data source; 42, block; 43, identifier:uuids_to_process; 44, comment:# Find which uuids to be processed exist in IDENTITIES_INDEX; 45, block; 46, call; 47, call; 48, identifier:target_uuids; 49, expression_statement; 50, expression_statement; 51, expression_statement; 52, if_statement; 53, return_statement; 54, identifier:target_uuids; 55, expression_statement; 56, expression_statement; 57, for_statement; 58, return_statement; 59, identifier:unique_ident; 60, identifier:data_sources; 61, expression_statement; 62, expression_statement; 63, for_statement; 64, return_statement; 65, identifier:unique_ident; 66, identifier:data_sources; 67, expression_statement; 68, expression_statement; 69, for_statement; 70, return_statement; 71, identifier:deleted_unique_identities; 72, integer:0; 73, identifier:deleted_identities; 74, integer:0; 75, identifier:uuids_to_process; 76, list; 77, attribute; 78, argument_list; 79, if_statement; 80, comment:# Remove the identities of non active data source for a given unique identity; 81, expression_statement; 82, comment:# Process only the unique identities that include the current data source, since; 83, comment:# it may be that unique identities in other data source have not been; 84, comment:# added yet to IDENTITIES_INDEX; 85, if_statement; 86, comment:# Add the uuid to the list to check its existence in the IDENTITIES_INDEX; 87, expression_statement; 88, comment:# Process the uuids in block of SIZE_SCROLL_IDENTITIES_INDEX; 89, if_statement; 90, comment:# Find which uuids to be processed exist in IDENTITIES_INDEX; 91, expression_statement; 92, expression_statement; 93, comment:# Find the uuids which exist in SortingHat but not in IDENTITIES_INDEX; 94, expression_statement; 95, comment:# Delete the orphan uuids from SortingHat; 96, expression_statement; 97, comment:# Reset the list; 98, expression_statement; 99, expression_statement; 100, expression_statement; 101, comment:# Find the uuids which exist in SortingHat but not in IDENTITIES_INDEX; 102, expression_statement; 103, comment:# Delete the orphan uuids from SortingHat; 104, expression_statement; 105, attribute; 106, argument_list; 107, attribute; 108, argument_list; 109, comment:"""Find a set of uuids in IDENTITIES_INDEX and return them if exist.
  
          :param target_uuids: target uuids
          """; 110, assignment; 111, assignment; 112, comparison_operator:page['hits']['total'] != 0; 113, block; 114, identifier:hits; 115, comment:"""Delete a list of uuids from SortingHat.
  
          :param target_uuids: uuids to be deleted
          """; 116, assignment; 117, identifier:uuid; 118, identifier:target_uuids; 119, block; 120, identifier:count; 121, comment:"""Remove the identities in non active data sources.
  
          :param unique_ident: unique identity object
          :param data_sources: target data sources
          """; 122, assignment; 123, identifier:ident; 124, attribute; 125, block; 126, identifier:count; 127, comment:"""Check if a unique identity has identities in a set of data sources.
  
          :param unique_ident: unique identity object
          :param data_sources: target data sources
          """; 128, assignment; 129, identifier:ident; 130, attribute; 131, block; 132, identifier:in_active; 133, identifier:SortingHat; 134, identifier:unique_identities; 135, identifier:sortinghat_db; 136, not_operator; 137, block; 138, augmented_assignment; 139, not_operator; 140, block; 141, call; 142, comparison_operator:len(uuids_to_process) != SIZE_SCROLL_IDENTITIES_INDEX; 143, block; 144, assignment; 145, assignment; 146, assignment; 147, augmented_assignment; 148, assignment; 149, assignment; 150, assignment; 151, assignment; 152, augmented_assignment; 153, identifier:logger; 154, identifier:debug; 155, string:"[identities retention] Total orphan unique identities deleted from SH: %i"; 156, identifier:deleted_unique_identities; 157, identifier:logger; 158, identifier:debug; 159, string:"[identities retention] Total identities in non-active data sources deleted from SH: %i"; 160, identifier:deleted_identities; 161, identifier:page; 162, call; 163, identifier:hits; 164, list; 165, subscript; 166, integer:0; 167, expression_statement; 168, identifier:count; 169, integer:0; 170, expression_statement; 171, expression_statement; 172, identifier:count; 173, integer:0; 174, identifier:unique_ident; 175, identifier:identities; 176, if_statement; 177, identifier:in_active; 178, False; 179, identifier:unique_ident; 180, identifier:identities; 181, if_statement; 182, call; 183, expression_statement; 184, continue_statement; 185, identifier:deleted_identities; 186, call; 187, call; 188, continue_statement; 189, attribute; 190, argument_list; 191, call; 192, identifier:SIZE_SCROLL_IDENTITIES_INDEX; 193, continue_statement; 194, identifier:results; 195, call; 196, identifier:uuids_found; 197, list_comprehension; 198, identifier:orphan_uuids; 199, binary_operator:set(uuids_to_process) - set(uuids_found); 200, identifier:deleted_unique_identities; 201, call; 202, identifier:uuids_to_process; 203, list; 204, identifier:results; 205, call; 206, identifier:uuids_found; 207, list_comprehension; 208, identifier:orphan_uuids; 209, binary_operator:set(uuids_to_process) - set(uuids_found); 210, identifier:deleted_unique_identities; 211, call; 212, attribute; 213, argument_list; 214, subscript; 215, string; 216, assignment; 217, assignment; 218, assignment; 219, comparison_operator:ident.source not in data_sources; 220, block; 221, comparison_operator:ident.source in data_sources; 222, block; 223, identifier:has_identities_in_data_sources; 224, argument_list; 225, augmented_assignment; 226, identifier:delete_identities; 227, argument_list; 228, identifier:has_identities_in_data_sources; 229, argument_list; 230, identifier:uuids_to_process; 231, identifier:append; 232, attribute; 233, identifier:len; 234, argument_list; 235, identifier:get_uuids_in_index; 236, argument_list; 237, subscript; 238, for_in_clause; 239, call; 240, call; 241, identifier:delete_unique_identities; 242, argument_list; 243, identifier:get_uuids_in_index; 244, argument_list; 245, subscript; 246, for_in_clause; 247, call; 248, call; 249, identifier:delete_unique_identities; 250, argument_list; 251, identifier:es; 252, identifier:search; 253, keyword_argument; 254, keyword_argument; 255, keyword_argument; 256, keyword_argument; 257, identifier:page; 258, string; 259, string_content:total; 260, identifier:hits; 261, subscript; 262, identifier:success; 263, call; 264, identifier:count; 265, conditional_expression:count + 1 if success else count; 266, attribute; 267, identifier:data_sources; 268, expression_statement; 269, expression_statement; 270, attribute; 271, identifier:data_sources; 272, expression_statement; 273, break_statement; 274, identifier:unique_identity; 275, identifier:active_data_sources; 276, identifier:deleted_unique_identities; 277, call; 278, identifier:unique_identity; 279, identifier:active_data_sources; 280, identifier:unique_identity; 281, list; 282, identifier:unique_identity; 283, identifier:uuid; 284, identifier:uuids_to_process; 285, identifier:uuids_to_process; 286, subscript; 287, string; 288, identifier:item; 289, identifier:results; 290, identifier:set; 291, argument_list; 292, identifier:set; 293, argument_list; 294, identifier:orphan_uuids; 295, identifier:uuids_to_process; 296, subscript; 297, string; 298, identifier:item; 299, identifier:results; 300, identifier:set; 301, argument_list; 302, identifier:set; 303, argument_list; 304, identifier:orphan_uuids; 305, identifier:index; 306, identifier:IDENTITIES_INDEX; 307, identifier:scroll; 308, string:"360m"; 309, identifier:size; 310, identifier:SIZE_SCROLL_IDENTITIES_INDEX; 311, identifier:body; 312, dictionary; 313, string_content:hits; 314, subscript; 315, string; 316, attribute; 317, argument_list; 318, binary_operator:count + 1; 319, identifier:success; 320, identifier:count; 321, identifier:ident; 322, identifier:source; 323, assignment; 324, assignment; 325, identifier:ident; 326, identifier:source; 327, assignment; 328, identifier:delete_unique_identities; 329, argument_list; 330, identifier:current_data_source; 331, identifier:item; 332, string; 333, string_content:sh_uuid; 334, identifier:uuids_to_process; 335, identifier:uuids_found; 336, identifier:item; 337, string; 338, string_content:sh_uuid; 339, identifier:uuids_to_process; 340, identifier:uuids_found; 341, pair; 342, identifier:page; 343, string; 344, string_content:hits; 345, identifier:SortingHat; 346, identifier:remove_unique_identity; 347, identifier:sortinghat_db; 348, identifier:uuid; 349, identifier:count; 350, integer:1; 351, identifier:success; 352, call; 353, identifier:count; 354, conditional_expression:count + 1 if success else count; 355, identifier:in_active; 356, True; 357, list; 358, string_content:_source; 359, string_content:_source; 360, string:"query"; 361, dictionary; 362, string_content:hits; 363, attribute; 364, argument_list; 365, binary_operator:count + 1; 366, identifier:success; 367, identifier:count; 368, attribute; 369, pair; 370, identifier:SortingHat; 371, identifier:remove_identity; 372, identifier:sortinghat_db; 373, attribute; 374, identifier:count; 375, integer:1; 376, identifier:unique_identity; 377, identifier:uuid; 378, string:"bool"; 379, dictionary; 380, identifier:ident; 381, identifier:id; 382, pair; 383, string:"filter"; 384, list; 385, dictionary; 386, pair; 387, string:"terms"; 388, dictionary; 389, pair; 390, string:"sh_uuid"; 391, identifier:target_uuids
code: def delete_orphan_unique_identities(es, sortinghat_db, current_data_source, active_data_sources):
      """Delete all unique identities which appear in SortingHat, but not in the IDENTITIES_INDEX.
  
      :param es: ElasticSearchDSL object
      :param sortinghat_db: instance of the SortingHat database
      :param current_data_source: current data source
      :param active_data_sources: list of active data sources
      """
      def get_uuids_in_index(target_uuids):
          """Find a set of uuids in IDENTITIES_INDEX and return them if exist.
  
          :param target_uuids: target uuids
          """
          page = es.search(
              index=IDENTITIES_INDEX,
              scroll="360m",
              size=SIZE_SCROLL_IDENTITIES_INDEX,
              body={
                  "query": {
                      "bool": {
                          "filter": [
                              {
                                  "terms": {
                                      "sh_uuid": target_uuids
                                  }
                              }
                          ]
                      }
                  }
              }
          )
  
          hits = []
          if page['hits']['total'] != 0:
              hits = page['hits']['hits']
  
          return hits
  
      def delete_unique_identities(target_uuids):
          """Delete a list of uuids from SortingHat.
  
          :param target_uuids: uuids to be deleted
          """
          count = 0
  
          for uuid in target_uuids:
              success = SortingHat.remove_unique_identity(sortinghat_db, uuid)
              count = count + 1 if success else count
  
          return count
  
      def delete_identities(unique_ident, data_sources):
          """Remove the identities in non active data sources.
  
          :param unique_ident: unique identity object
          :param data_sources: target data sources
          """
          count = 0
          for ident in unique_ident.identities:
              if ident.source not in data_sources:
                  success = SortingHat.remove_identity(sortinghat_db, ident.id)
                  count = count + 1 if success else count
  
          return count
  
      def has_identities_in_data_sources(unique_ident, data_sources):
          """Check if a unique identity has identities in a set of data sources.
  
          :param unique_ident: unique identity object
          :param data_sources: target data sources
          """
          in_active = False
          for ident in unique_ident.identities:
              if ident.source in data_sources:
                  in_active = True
                  break
  
          return in_active
  
      deleted_unique_identities = 0
      deleted_identities = 0
      uuids_to_process = []
  
      # Collect all unique identities
      for unique_identity in SortingHat.unique_identities(sortinghat_db):
  
          # Remove a unique identity if all its identities are in non active data source
          if not has_identities_in_data_sources(unique_identity, active_data_sources):
              deleted_unique_identities += delete_unique_identities([unique_identity.uuid])
              continue
  
          # Remove the identities of non active data source for a given unique identity
          deleted_identities += delete_identities(unique_identity, active_data_sources)
  
          # Process only the unique identities that include the current data source, since
          # it may be that unique identities in other data source have not been
          # added yet to IDENTITIES_INDEX
          if not has_identities_in_data_sources(unique_identity, [current_data_source]):
              continue
  
          # Add the uuid to the list to check its existence in the IDENTITIES_INDEX
          uuids_to_process.append(unique_identity.uuid)
  
          # Process the uuids in block of SIZE_SCROLL_IDENTITIES_INDEX
          if len(uuids_to_process) != SIZE_SCROLL_IDENTITIES_INDEX:
              continue
  
          # Find which uuids to be processed exist in IDENTITIES_INDEX
          results = get_uuids_in_index(uuids_to_process)
          uuids_found = [item['_source']['sh_uuid'] for item in results]
  
          # Find the uuids which exist in SortingHat but not in IDENTITIES_INDEX
          orphan_uuids = set(uuids_to_process) - set(uuids_found)
          # Delete the orphan uuids from SortingHat
          deleted_unique_identities += delete_unique_identities(orphan_uuids)
          # Reset the list
          uuids_to_process = []
  
      # Check that no uuids have been left to process
      if uuids_to_process:
          # Find which uuids to be processed exist in IDENTITIES_INDEX
          results = get_uuids_in_index(uuids_to_process)
          uuids_found = [item['_source']['sh_uuid'] for item in results]
  
          # Find the uuids which exist in SortingHat but not in IDENTITIES_INDEX
          orphan_uuids = set(uuids_to_process) - set(uuids_found)
  
          # Delete the orphan uuids from SortingHat
          deleted_unique_identities += delete_unique_identities(orphan_uuids)
  
      logger.debug("[identities retention] Total orphan unique identities deleted from SH: %i",
                   deleted_unique_identities)
      logger.debug("[identities retention] Total identities in non-active data sources deleted from SH: %i",
                   deleted_identities)