Skip to content

Commit

Permalink
feat: add a get_work_item function to get a single work item with all…
Browse files Browse the repository at this point in the history
… fields
  • Loading branch information
micha91 committed Jan 16, 2024
1 parent 01e3043 commit 55f706b
Show file tree
Hide file tree
Showing 6 changed files with 408 additions and 67 deletions.
9 changes: 9 additions & 0 deletions polarion_rest_api_client/base_client.py
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
157 changes: 90 additions & 67 deletions polarion_rest_api_client/client.py
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,102 @@ 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."""
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 = []
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_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,
)
return work_item_obj

def get_document(
self,
space_id: str,
Expand Down
1 change: 1 addition & 0 deletions tests/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@
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_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

0 comments on commit 55f706b

Please sign in to comment.