refactor: Implement sync client without event loop #89
Conversation
rossum_api/elis_api_client_sync.py
Outdated
| ) | ||
| """https://elis.rossum.ai/api/docs/#search-for-annotations.""" | ||
| validate_search_params(query, query_string) | ||
| json_payload = build_search_params(query, query_string) |
There was a problem hiding this comment.
Nitpick: I would rename this variable to search_params
rossum_api/elis_api_client_sync.py
Outdated
| return self._run_coroutine( | ||
| self.elis_api_client.poll_annotation_until_imported(annotation_id, **poll_kwargs) | ||
| ) | ||
| return self.poll_task(task_id, lambda a: a.status == TaskStatus.SUCCEEDED, sleep_s) |
There was a problem hiding this comment.
Perhaps it would be worth extracting this lambda to a function in domain logic like we did with is_annotation_imported
rossum_api/elis_api_client_sync.py
Outdated
|
|
||
| def delete_annotation(self, annotation_id: int) -> None: | ||
| """https://elis.rossum.ai/api/docs/#switch-to-deleted""" | ||
| return self._run_coroutine(self.elis_api_client.delete_annotation(annotation_id)) | ||
| self.internal_client.request( | ||
| "POST", url=f"{Resource.Annotation.value}/{annotation_id}/delete" |
There was a problem hiding this comment.
I would create little functions for all these URLs in domain_logic.urls for re-usability. They're kind of dumb but it's imho worth it.
rossum_api/internal_sync_client.py
Outdated
| raise AlwaysRetry() | ||
| self._raise_for_status(response) | ||
| for chunk in response.iter_bytes(): | ||
| yield chunk |
There was a problem hiding this comment.
Nitpick: We're not in async anymore, we can do yield from response.iter_bytes() or probably even return response.iter_bytes()
rossum_api/internal_sync_client.py
Outdated
| json, | ||
| **filters, | ||
| ): | ||
| yield result |
There was a problem hiding this comment.
Same here: return self.fetch_resources_by_url(... should work.
rossum_api/internal_sync_client.py
Outdated
| ) -> Iterator[dict[str, Any]]: | ||
| query_params = build_pagination_params(ordering) | ||
| query_params.update(build_sideload_params(sideloads, content_schema_ids)) | ||
| query_params.update(**filters) |
There was a problem hiding this comment.
No strong opinion but we could create a higher level function build_fetch_resources_params(ordering, sideloads, content_schema_ids)
rossum_api/elis_api_client_sync.py
Outdated
|
|
||
| if sideloads: | ||
| self.internal_client.sideload(annotation_response, sideloads) | ||
| return self._deserializer(resource, annotation_response) |
There was a problem hiding this comment.
I remember we were not happy about this part during the hackaton. I think we could simplify this at the expense of 1 extra API call at the end (fetching the annotation once more instead of just the sideloads).
annotation = self.retrieve_annotation(annotation_id)
while not predicate(annotation):
time.sleep(sleep_s)
annotation = self.retrieve_annotation(annotation_id)
return self.retrieve_annotation(annotation_id, sideloads)
Outcome of our Xmax hackathon 2024