Combine the results from one page with the existing results for a given query. Parameters: page_results: The results for the current page.
(self, page_results: dict)
| 49 | return asdict(self) |
| 50 | |
| 51 | def update(self, page_results: dict) -> None: |
| 52 | """Combine the results from one page with the existing results for a given query. |
| 53 | |
| 54 | Parameters: |
| 55 | page_results: The results for the current page. |
| 56 | |
| 57 | """ |
| 58 | |
| 59 | if "data" in page_results: |
| 60 | # If the `data` field is a list, add it to our existing results. |
| 61 | # Otherwise, as is the case for `info` endpoints, `data` is a dictionary (or null) |
| 62 | # and should be added as the only `data` field for these results. |
| 63 | if isinstance(page_results["data"], list): |
| 64 | self.data.extend(page_results["data"]) # type: ignore[union-attr] |
| 65 | elif not self.data: |
| 66 | self.data = page_results["data"] |
| 67 | else: |
| 68 | raise RuntimeError( |
| 69 | "Not overwriting old `data` field in `QueryResults`." |
| 70 | ) |
| 71 | |
| 72 | if "errors" in page_results: |
| 73 | self.errors.extend(page_results["errors"]) |
| 74 | |
| 75 | # Combine meta/links fields across all pages in a sensible way, i.e., |
| 76 | # if we really reached the last page of results, then make sure `links->next` |
| 77 | # is null in the final response, and make sure `meta->more_data_available` is None or False. |
| 78 | keys_to_filter = { |
| 79 | "links": ("next", "prev"), |
| 80 | "meta": ("query", "more_data_available"), |
| 81 | } |
| 82 | for top_level_key in keys_to_filter: |
| 83 | if top_level_key not in page_results: |
| 84 | page_results[top_level_key] = {} |
| 85 | for k in keys_to_filter[top_level_key]: |
| 86 | if k not in page_results[top_level_key]: |
| 87 | page_results[top_level_key][k] = None |
| 88 | getattr(self, top_level_key).update( |
| 89 | {k: page_results[top_level_key][k] for k in page_results[top_level_key]} |
| 90 | ) |
| 91 | |
| 92 | # Only add new unique entries to the included list |
| 93 | for d in page_results.get("included", []): |
| 94 | typed_id = f"{d['type']}/{d['id']}" |
| 95 | if typed_id not in self.included_index: |
| 96 | self.included_index.add(typed_id) |
| 97 | self.included.append(d) |
| 98 | |
| 99 | |
| 100 | class OptimadeClientProgress(Progress): |