refactor: Implement sync client without event loop

.pre-commit-config.yaml

@@ -1,13 +1,12 @@
 repos:
-  - repo: https://github.com/ambv/black
-    rev: 22.3.0
-    hooks:
-      - id: black
-  - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.0.259
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.8.2
     hooks:
       - id: ruff
-        args: [--fix, --exit-non-zero-on-fix]
+        name: sort imports with ruff
+        args: [--select, I, --fix]
+      - id: ruff-format
+        name: format with ruff
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v0.981
     hooks:

pyproject.toml

@@ -13,6 +13,7 @@ dependencies = [
     "dacite",
     "httpx",
     "tenacity",
+    "inflect",
 ]
 
 [project.optional-dependencies]
@@ -24,6 +25,7 @@ tests = [
   "pytest-cov",
   "ruff",
   "types-aiofiles",
+  "pre-commit",
 ]
 
 [tools.setuptools]

rossum_api/domain_logic/annotations.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from rossum_api.models import Annotation
+
+if TYPE_CHECKING:
+    from typing import Sequence
+
+
+def validate_list_annotations_params(
+    sideloads: Sequence[str] = (),
+    content_schema_ids: Sequence[str] = (),
+) -> None:
+    """Validate parameters to list_annotations request."""
+    if sideloads and "content" in sideloads and not content_schema_ids:
+        raise ValueError(
+            'When content sideloading is requested, "content_schema_ids" must be provided'
+        )
+
+
+def get_http_method_for_annotation_export(**filters) -> str:
+    """to_status filter requires a different HTTP method.
+
+    https://elis.rossum.ai/api/docs/#export-annotations
+    """
+    if "to_status" in filters:
+        return "POST"
+    return "GET"
+
+
+def is_annotation_imported(annotation: Annotation) -> bool:
+    return annotation.status not in ("importing", "created")

rossum_api/domain_logic/documents.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Optional
+
+import httpx
+
+
+def build_create_document_params(
+    file_name: str,
+    file_data: bytes,
+    metadata: Optional[dict[str, Any]],
+    parent: Optional[str],
+) -> dict[str, Any]:
+    metadata = metadata or {}
+    files: httpx._types.RequestFiles = {
+        "content": (file_name, file_data),
+        "metadata": ("", json.dumps(metadata).encode("utf-8")),
+    }
+    if parent:
+        files["parent"] = ("", parent)
+    return files

rossum_api/domain_logic/pagination.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Sequence
+
+DEFAULT_PAGE_SIZE = 100
+
+
+def build_pagination_params(ordering: Sequence[str], page_size: int = DEFAULT_PAGE_SIZE) -> dict:
+    """Build params used for fetching paginated resources."""
+    return {
+        "page_size": page_size,
+        "ordering": ",".join(ordering),
+    }

rossum_api/domain_logic/retry.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+import httpx
+
+RETRIED_HTTP_CODES = (408, 429, 500, 502, 503, 504)
+
+
+class AlwaysRetry(Exception):
+    pass
+
+
+def should_retry(exc: BaseException) -> bool:
+    if isinstance(exc, (AlwaysRetry, httpx.RequestError)):
+        return True
+    if isinstance(exc, httpx.HTTPStatusError):
+        return exc.response.status_code in RETRIED_HTTP_CODES
+    return False

rossum_api/domain_logic/search.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from typing import Any, Optional
+
+
+def validate_search_params(
+    query: Optional[dict] = None,
+    query_string: Optional[dict] = None,
+):
+    if not query and not query_string:
+        raise ValueError("Either query or query_string must be provided")
+
+
+def build_search_params(
+    query: Optional[dict] = None,
+    query_string: Optional[dict] = None,
+) -> dict[str, Any]:
+    json_payload = {}
+    if query:
+        json_payload["query"] = query
+    if query_string:
+        json_payload["query_string"] = query_string
+    return json_payload

rossum_api/domain_logic/sideloads.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import itertools
+from typing import TYPE_CHECKING
+
+from rossum_api.domain_logic.urls import (
+    parse_annotation_id_from_datapoint_url,
+    parse_resource_id_from_url,
+)
+from rossum_api.utils import to_singular
+
+if TYPE_CHECKING:
+    from typing import Any, Sequence, Union
+
+
+def _group_sideloads_by_annotation_id(
+    sideloads: Sequence[str], response_data: dict[str, Any]
+) -> dict[str, dict[int, Union[dict, list]]]:
+    sideloads_by_id: dict[str, dict[int, Union[dict, list]]] = {}
+    for sideload in sideloads:
+        if sideload == "content":
+            # Datapoints from all annotations are present in response data, we have to construct
+            # content (list of datapoints) for each annotation.
+            def get_annotation_id(datapoint: dict[str, Any]) -> int:
+                return parse_annotation_id_from_datapoint_url(datapoint["url"])
+
+            sideloads_by_id[sideload] = {
+                k: list(v)
+                for k, v in itertools.groupby(
+                    sorted(response_data[sideload], key=get_annotation_id),
+                    key=get_annotation_id,
+                )
+            }
+        else:
+            sideloads_by_id[sideload] = {s["id"]: s for s in response_data[sideload]}
+    return sideloads_by_id
+
+
+def embed_sideloads(response_data, sideloads: Sequence[str]) -> None:
+    """Put sideloads into the response data."""
+    sideloads_by_id = _group_sideloads_by_annotation_id(sideloads, response_data)
+    for result, sideload in itertools.product(response_data["results"], sideloads):
+        sideload_name = to_singular(sideload)
+        url = result[sideload_name]
+        if url is None:
+            continue
+        sideload_id = parse_resource_id_from_url(url)
+
+        result[sideload_name] = sideloads_by_id[sideload].get(
+            sideload_id, []
+        )  # `content` can have 0 datapoints, use [] default value in this case
+
+
+def build_sideload_params(sideloads: Sequence[str], content_schema_ids: Sequence[str]) -> dict:
+    """Build params used for sideloading."""
+    return {
+        "sideload": ",".join(sideloads),
+        "content.schema_id": ",".join(content_schema_ids),
+    }

rossum_api/domain_logic/upload.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from typing import Any, BinaryIO, Optional
+
+
+def build_upload_files(
+    fp: BinaryIO,
+    filename: str,
+    values: Optional[dict[str, Any]] = None,
+    metadata: Optional[dict[str, Any]] = None,
+) -> dict[str, Any]:
+    """Build request files for the upload endpoint."""
+    files = {"content": (filename, fp.read(), "application/octet-stream")}
+
+    # Filename of values and metadata must be "", otherwise Elis API returns HTTP 400 with body
+    # "Value must be valid JSON."
+    if values is not None:
+        files["values"] = ("", json.dumps(values).encode("utf-8"), "application/json")
+    if metadata is not None:
+        files["metadata"] = ("", json.dumps(metadata).encode("utf-8"), "application/json")
+
+    return files

rossum_api/domain_logic/urls.py

@@ -3,6 +3,8 @@
 import re
 from typing import TYPE_CHECKING
 
+from rossum_api.api_client import Resource
+
 if TYPE_CHECKING:
     from rossum_api.models import Resource
 

rossum_api/dtos.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+import dataclasses
+
+
+@dataclasses.dataclass
+class Token:
+    token: str
+
+
+@dataclasses.dataclass
+class UserCredentials:
+    username: str
+    password: str

rossum_api/elis_api_client.py

@@ -573,7 +573,7 @@ async def list_all_email_templates(
         self,
         ordering: Sequence[str] = (),
         **filters: Any,
-    ) -> AsyncIterator[Connector]:
+    ) -> AsyncIterator[EmailTemplate]:
         """https://elis.rossum.ai/api/docs/#list-all-email-templates."""
         async for c in self._http_client.fetch_all(Resource.EmailTemplate, ordering, **filters):
             yield self._deserializer(Resource.EmailTemplate, c)