get_documents call

Author: dahbar

polarion_rest_api_client/base_client.py

@@ -9,6 +9,7 @@
 from polarion_rest_api_client import data_models as dm
 
 WorkItemType = t.TypeVar("WorkItemType", bound=dm.WorkItem)
+DocumentType = t.TypeVar("DocumentType", bound=dm.Document)
 
 
 class DefaultFields:
@@ -17,6 +18,7 @@ class DefaultFields:
     _workitems: str = "@basic"
     _linkedworkitems: str = "id,role,suspect"
     _workitem_attachments: str = "@basic"
+    _documents: str = "@basic"
 
     @property
     def workitems(self):
@@ -45,11 +47,23 @@ def workitem_attachments(self):
     def workitem_attachments(self, value):
         self._workitem_attachments = value
 
+    @property
+    def documents(self):
+        """Return the fields dict for document."""
+        return {"documents": self._documents}
+
+    @documents.setter
+    def documents(self, value):
+        self._documents = value
+
     @property
     def all_types(self):
         """Return all fields dicts merged together."""
         return (
-            self.workitem_attachments | self.workitems | self.linkedworkitems
+            self.workitem_attachments
+            | self.workitems
+            | self.linkedworkitems
+            | self.documents
         )
 
 

polarion_rest_api_client/client.py

@@ -35,6 +35,7 @@
     patch_work_item,
     post_work_items,
 )
+from polarion_rest_api_client.open_api_client.api.documents import get_document
 
 logger = logging.getLogger(__name__)
 
@@ -611,16 +612,20 @@ def get_work_items(
                         self._work_item(
                             work_item_id,
                             unset_str_builder(work_item.attributes.title),
-                            unset_str_builder(
-                                work_item.attributes.description.type
-                            )
-                            if work_item.attributes.description
-                            else None,
-                            unset_str_builder(
-                                work_item.attributes.description.value
-                            )
-                            if work_item.attributes.description
-                            else None,
+                            (
+                                unset_str_builder(
+                                    work_item.attributes.description.type
+                                )
+                                if work_item.attributes.description
+                                else None
+                            ),
+                            (
+                                unset_str_builder(
+                                    work_item.attributes.description.value
+                                )
+                                if work_item.attributes.description
+                                else None
+                            ),
                             unset_str_builder(work_item.attributes.type),
                             unset_str_builder(work_item.attributes.status),
                             work_item.attributes.additional_properties,
@@ -635,6 +640,66 @@ def get_work_items(
 
         return work_items, next_page
 
+    def get_document(
+        self,
+        space_id: str,
+        document_name: str,
+        fields: dict[str, str] | None = None,
+        include: str | None = None,
+        revision: str | None = None,
+        retry: bool = True,
+    ) -> base_client.DocumentType:
+        """Return the document with the given document_name and space_id."""
+
+        if " " in space_id or " " in document_name:
+            space_id = space_id.replace(" ", "%20")
+            document_name = document_name.replace(" ", "%20")
+        if fields is None:
+            fields = self.default_fields.documents
+
+        sparse_fields = _build_sparse_fields(fields)
+        response = get_document.sync_detailed(
+            self.project_id,
+            space_id,
+            document_name,
+            client=self.client,
+            fields=sparse_fields,
+            include=include,
+            revision=revision,
+        )
+
+        if not self._check_response(response, not retry) and retry:
+            sleep_random_time()
+            return self.get_work_items(
+                space_id, document_name, fields, include, revision, False
+            )
+
+        document_response = response.parsed
+
+        if isinstance(
+            document_response, api_models.DocumentsSingleGetResponse
+        ) and (data := document_response.data):
+            if not getattr(data.meta, "errors", []):
+                assert (attributes := data.attributes)
+                assert isinstance(data.id, str)
+
+                document: base_client.DocumentType = dm.Document(
+                    id=data.id,
+                    module_folder=unset_str_builder(attributes.module_folder),
+                    module_name=unset_str_builder(attributes.module_name),
+                    type=unset_str_builder(attributes.type),
+                    status=unset_str_builder(attributes.status),
+                    home_page_content={
+                        "type": attributes.home_page_content.type,
+                        "value": attributes.home_page_content.value,
+                    }
+                    if not isinstance(
+                        attributes.home_page_content, oa_types.Unset
+                    )
+                    else None,
+                )
+        return document
+
     def create_work_items(self, work_items: list[base_client.WorkItemType]):
         """Create the given list of work items."""
         current_batch = api_models.WorkitemsListPostRequest([])

polarion_rest_api_client/data_models.py

@@ -151,3 +151,74 @@ class WorkItemAttachment:
     content_bytes: bytes | None = None
     mime_type: str | None = None
     file_name: str | None = None
+
+
+class Document:
+    """A data class containing all relevant data of a Polarion WorkItem."""
+
+    id: str | None = None
+    module_folder: str | None = None
+    module_name: str | None = None
+    type: str | None = None
+    status: str | None = None
+    home_page_content: dict | None = None
+    _checksum: str | None = None
+
+    def __init__(
+        self,
+        id: str | None = None,
+        module_folder: str | None = None,
+        module_name: str | None = None,
+        type: str | None = None,
+        status: str | None = None,
+        home_page_content: dict | None = None,
+    ):
+        self.id = id
+        self.module_folder = module_folder
+        self.module_name = module_name
+        self.type = type
+        self.status = status
+        self.home_page_content = home_page_content
+        self._checksum = None
+
+    def __eq__(self, other: object) -> bool:
+        """Compare only Document attributes."""
+        if not isinstance(other, Document):
+            return NotImplemented
+        if self.get_current_checksum() is None:
+            self.calculate_checksum()
+        if other.get_current_checksum() is None:
+            other.calculate_checksum()
+
+        return self.get_current_checksum() == other.get_current_checksum()
+
+    def to_dict(self) -> dict[str, t.Any]:
+        """Return the content of the Document as dictionary."""
+        return {
+            "id": self.id,
+            "module_folder": self.module_folder,
+            "module_name": self.module_name,
+            "type": self.type,
+            "status": self.status,
+            "home_page_content": self.home_page_content,
+            "checksum": self._checksum,
+        }
+
+    def calculate_checksum(self) -> str:
+        """Calculate and return a checksum for this Document.
+
+        In addition, the checksum will be written to self._checksum.
+        """
+        data = self.to_dict()
+        del data["checksum"]
+        del data["id"]
+
+        data = dict(sorted(data.items()))
+
+        converted = json.dumps(data).encode("utf8")
+        self._checksum = hashlib.sha256(converted).hexdigest()
+        return self._checksum
+
+    def get_current_checksum(self) -> str | None:
+        """Return the checksum currently set without calculation."""
+        return self._checksum

tests/__init__.py

@@ -59,7 +59,7 @@
     TEST_RESPONSES / "workitems_next_page_error.json"
 )
 TEST_WI_NEXT_PAGE_RESPONSE = TEST_RESPONSES / "workitems_next_page.json"
-
+TEST_DOCUMENT_RESPONSE = TEST_RESPONSES / "get_document.json"
 TEST_ERROR_RESPONSE = TEST_RESPONSES / "error.json"
 TEST_PROJECT_RESPONSE_JSON = TEST_RESPONSES / "project.json"
 

tests/conftest.py

@@ -69,3 +69,15 @@ def fixture_dummy_work_item_attachment():
         "text/plain",
         "test.json",
     )
+
+
+@pytest.fixture(name="document")
+def fixture_dummy_document():
+    return polarion_api.Document(
+        "MyProjectId/MySpaceId/MyDocumentName",
+        "MySpaceId",
+        "MyDocumentName",
+        "standardSpecification",
+        "open",
+        {"type": "text/html", "value": "<h1>My text value</h1>"},
+    )

tests/data/mock_api_responses/get_document.json

@@ -0,0 +1,36 @@
+{
+    "data": {
+        "attributes": {
+            "autoSuspect": {},
+            "branchedWithQuery": {},
+            "derivedFromLinkRole": {},
+            "created": "1970-01-01T00:00:00Z",
+            "homePageContent": {
+                "type": "text/html",
+                "value": "<h1>My text value</h1>"
+            },
+            "moduleFolder": "MySpaceId",
+            "moduleName": "MyDocumentName",
+            "outlineNumbering": {},
+            "renderingLayouts": [],
+            "status": "open",
+            "structureLinkRole": "parent",
+            "title": "MyDocumentName",
+            "type": "standardSpecification",
+            "updated": "1970-01-01T00:00:00Z",
+            "usesOutlineNumbering": true
+        },
+        "id": "MyProjectId/MySpaceId/MyDocumentName",
+        "links": {
+            "self": "server-host-name/application-path/projects/MyProjectId/spaces/MySpaceId/documents/MyDocumentName"
+        },
+        "meta": {},
+        "relationships": {},
+        "revision": {},
+        "type": "documents"
+    },
+    "included": [],
+    "links": {
+        "self": "server-host-name/application-path/projects/MyProjectId/spaces/MySpaceId/documents/MyDocumentName"
+    }
+}

tests/data/mock_api_responses/get_document.json.license

@@ -0,0 +1,2 @@
+Copyright DB Netz AG and contributors
+SPDX-License-Identifier: Apache-2.0

tests/test_client_documents.py

@@ -0,0 +1,40 @@
+# Copyright DB Netz AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import copy
+import json
+
+import pytest
+import pytest_httpx
+import pytest_mock as mock
+
+import polarion_rest_api_client as polarion_api
+from polarion_rest_api_client.open_api_client import models as api_models
+from tests import TEST_DOCUMENT_RESPONSE
+
+
+def test_get_document_with_all_fields(
+    client: polarion_api.OpenAPIPolarionProjectClient,
+    httpx_mock: pytest_httpx.HTTPXMock,
+):
+    with open(TEST_DOCUMENT_RESPONSE, encoding="utf8") as f:
+        httpx_mock.add_response(json=json.load(f))
+
+    client.default_fields.documents = "@all"
+
+    document = client.get_document("MySpaceId", "MyDocumentName")
+
+    reqs = httpx_mock.get_requests()
+    assert reqs[0].method == "GET"
+    assert isinstance(document, polarion_api.data_models.Document)
+    assert len(reqs) == 1
+    assert document == polarion_api.data_models.Document(
+        "MyProjectId/MySpaceId/MyDocumentName",
+        "MySpaceId",
+        "MyDocumentName",
+        "standardSpecification",
+        "open",
+        {"type": "text/html", "value": "<h1>My text value</h1>"},
+    )