Skip to content

feat: Implement get single work item and detect truncated relationships #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
Jan 17, 2024
Merged

feat: Implement get single work item and detect truncated relationships #22

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
55f706b
feat: add a get_work_item function to get a single work item with all…
micha91 Jan 16, 2024
da1870f
feat: add 2 new flags to the WorkItem Class telling the user, whether…
micha91 Jan 17, 2024
11d0da8
docs: Add a comment to explain, how the truncated flags are set if a …
micha91 Jan 17, 2024
3dac71b
refactor: Add changes from review
micha91 Jan 17, 2024
176471f
refactor: Add changes from review
micha91 Jan 17, 2024
79337c7
refactor: apply feedback from review
micha91 Jan 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions polarion_rest_api_client/base_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,6 +204,15 @@ def get_work_items(
"""
raise NotImplementedError

@abc.abstractmethod
def get_work_item(
self,
work_item_id: str,
retry: bool = True,
) -> WorkItemType | None:
"""Return one specific work item with all fields."""
raise NotImplementedError

def create_work_item(self, work_item: WorkItemType):
"""Create a single given work item."""
self.create_work_items([work_item])
Expand Down
174 changes: 107 additions & 67 deletions polarion_rest_api_client/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
)
from polarion_rest_api_client.open_api_client.api.work_items import (
delete_work_items,
get_work_item,
get_work_items,
patch_work_item,
post_work_items,
Expand Down Expand Up @@ -567,80 +568,119 @@ def get_work_items(
):
for work_item in work_items_response.data:
if not getattr(work_item.meta, "errors", []):
assert work_item.attributes
assert isinstance(work_item.id, str)
work_item_id = work_item.id.split("/")[-1]

work_item_links = []
work_item_attachments = []

if (
work_item.relationships
and work_item.relationships.linked_work_items
and work_item.relationships.linked_work_items.data
):
for (
link
) in work_item.relationships.linked_work_items.data:
work_item_links.append(
self._parse_work_item_link(
link.id,
link.additional_properties.get(
"suspect", False
),
work_item_id,
)
)

if (
work_item.relationships
and work_item.relationships.attachments
and work_item.relationships.attachments.data
):
for (
attachment
) in work_item.relationships.attachments.data:
assert attachment.id
work_item_attachments.append(
dm.WorkItemAttachment(
work_item_id,
attachment.id.split("/")[-1],
None, # title isn't provided
)
)

work_items.append(
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.type),
unset_str_builder(work_item.attributes.status),
work_item.attributes.additional_properties,
work_item_links,
work_item_attachments,
)
)
work_item_obj = self._generate_work_item(work_item)
work_items.append(work_item_obj)
next_page = isinstance(
work_items_response.links,
api_models.WorkitemsListGetResponseLinks,
) and bool(work_items_response.links.next_)

return work_items, next_page

def get_work_item(
self,
work_item_id: str,
retry: bool = True,
) -> base_client.WorkItemType | None:
"""Return one specific work item with all fields.

This also includes all linked work items and attachments. If
there are to many of these to get them in one request, the
truncated flags for linked_work_items and attachments will be
set to True.
"""
response = get_work_item.sync_detailed(
self.project_id,
work_item_id,
client=self.client,
fields=_build_sparse_fields(
{
"workitems": "@all",
"workitem_attachments": "@all",
"linkedworkitems": "@all",
}
),
)
if not self._check_response(response, not retry) and retry:
sleep_random_time()
return self.get_work_item(work_item_id, False)

if response.parsed and isinstance(
response.parsed.data, api_models.WorkitemsSingleGetResponseData
):
return self._generate_work_item(response.parsed.data)

return None

def _generate_work_item(
self,
work_item: api_models.WorkitemsListGetResponseDataItem
| api_models.WorkitemsSingleGetResponseData,
) -> base_client.WorkItemType:
assert work_item.attributes
assert isinstance(work_item.id, str)
work_item_id = work_item.id.split("/")[-1]
work_item_links = []
work_item_attachments = []

# We set both truncated flags to True and will only set them to False,
# if the corresponding fields were requested and returned completely
links_truncated = True
attachments_truncated = True
if work_item.relationships:
if links := work_item.relationships.linked_work_items:
if not links.meta or links.meta.total_count is oa_types.UNSET:
links_truncated = False

work_item_links = [
self._parse_work_item_link(
link.id,
link.additional_properties.get("suspect", False),
work_item_id,
)
for link in links.data or []
]

if attachments := work_item.relationships.attachments:
if (
not attachments.meta
or attachments.meta.total_count is oa_types.UNSET
):
attachments_truncated = False

work_item_attachments = [
dm.WorkItemAttachment(
work_item_id,
attachment.id.split("/")[-1],
None, # title isn't provided
)
for attachment in attachments.data or []
if attachment.id
]

work_item_obj = 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.type),
unset_str_builder(work_item.attributes.status),
work_item.attributes.additional_properties,
work_item_links,
work_item_attachments,
links_truncated,
attachments_truncated,
)
return work_item_obj

def get_document(
self,
space_id: str,
Expand Down
6 changes: 6 additions & 0 deletions polarion_rest_api_client/data_models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,9 @@ class WorkItem(BaseItem):
description: str | None = None
additional_attributes: dict[str, t.Any] = {}
linked_work_items: list[WorkItemLink] = []
linked_work_items_truncated: bool = False
attachments: list[WorkItemAttachment] = []
attachments_truncated: bool = False

def __init__(
self,
Expand All @@ -80,6 +82,8 @@ def __init__(
additional_attributes: dict[str, t.Any] | None = None,
linked_work_items: list[WorkItemLink] | None = None,
attachments: list[WorkItemAttachment] | None = None,
linked_work_items_truncated: bool = False,
attachments_truncated: bool = False,
**kwargs,
):
super().__init__(id, type, status)
Expand All @@ -89,6 +93,8 @@ def __init__(
self.additional_attributes = (additional_attributes or {}) | kwargs
self.linked_work_items = linked_work_items or []
self.attachments = attachments or []
self.linked_work_items_truncated = linked_work_items_truncated
self.attachments_truncated = attachments_truncated

def __getattribute__(self, item: str) -> t.Any:
"""Return all non WorkItem attributes from additional_properties."""
Expand Down
4 changes: 4 additions & 0 deletions tests/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,10 @@
TEST_RESPONSES / "workitems_next_page_error.json"
)
TEST_WI_NEXT_PAGE_RESPONSE = TEST_RESPONSES / "workitems_next_page.json"
TEST_WI_SINGLE_RESPONSE = TEST_RESPONSES / "get_work_item.json"
TEST_WI_NOT_TRUNCATED_RESPONSE = (
TEST_RESPONSES / "get_work_item_not_truncated.json"
)
TEST_DOCUMENT_RESPONSE = TEST_RESPONSES / "get_document.json"
TEST_ERROR_RESPONSE = TEST_RESPONSES / "error.json"
TEST_PROJECT_RESPONSE_JSON = TEST_RESPONSES / "project.json"
Expand Down
Loading