| 101 | |
| 102 | |
| 103 | class MultiServerWikidataQueryClient: |
| 104 | def __init__(self, urls: tp.List[str]): |
| 105 | self.clients = [WikidataQueryClient(url) for url in urls] |
| 106 | self.executor = ThreadPoolExecutor(max_workers=len(urls)) |
| 107 | # # test connections |
| 108 | # start_time = time.perf_counter() |
| 109 | # self.test_connections() |
| 110 | # end_time = time.perf_counter() |
| 111 | # print(f"Connection testing took {end_time - start_time} seconds") |
| 112 | |
| 113 | def test_connections(self): |
| 114 | def test_url(client): |
| 115 | try: |
| 116 | # Check if server provides the system.listMethods function. |
| 117 | client.server.system.listMethods() |
| 118 | return True |
| 119 | except Exception as e: |
| 120 | print(f"Failed to connect to {client.url}. Error: {str(e)}") |
| 121 | return False |
| 122 | |
| 123 | start_time = time.perf_counter() |
| 124 | futures = [ |
| 125 | self.executor.submit(test_url, client) for client in self.clients |
| 126 | ] |
| 127 | results = [f.result() for f in futures] |
| 128 | end_time = time.perf_counter() |
| 129 | # print(f"Testing connections took {end_time - start_time} seconds") |
| 130 | # Remove clients that failed to connect |
| 131 | self.clients = [ |
| 132 | client for client, result in zip(self.clients, results) if result |
| 133 | ] |
| 134 | if not self.clients: |
| 135 | raise Exception("Failed to connect to all URLs") |
| 136 | |
| 137 | def query_all(self, method, *args): |
| 138 | start_time = time.perf_counter() |
| 139 | futures = [ |
| 140 | self.executor.submit(getattr(client, method), *args) |
| 141 | for client in self.clients |
| 142 | ] |
| 143 | # Retrieve results and filter out 'Not Found!' |
| 144 | is_dict_return = method in [ |
| 145 | "get_all_relations_of_an_entity", |
| 146 | "get_tail_entities_given_head_and_relation", |
| 147 | ] |
| 148 | results = [f.result() for f in futures] |
| 149 | end_time = time.perf_counter() |
| 150 | # print(f"HTTP Queries took {end_time - start_time} seconds") |
| 151 | |
| 152 | start_time = time.perf_counter() |
| 153 | real_results = ( |
| 154 | set() if not is_dict_return else {"head": [], "tail": []} |
| 155 | ) |
| 156 | for res in results: |
| 157 | if isinstance(res, str) and res == "Not Found!": |
| 158 | continue |
| 159 | elif isinstance(res, tp.List): |
| 160 | if len(res) == 0: |
no outgoing calls
no test coverage detected