From a18e6c692dc81881d8f494e11bfba4b1a8367f1e Mon Sep 17 00:00:00 2001 From: Oleg A Date: Fri, 25 Aug 2023 22:06:34 +0300 Subject: [PATCH] refactor: add api categories --- yatracker/tracker/api.py | 235 +---------------- yatracker/tracker/categories/__init__.py | 9 + .../tracker/categories/attached_files.py | 0 .../tracker/categories/bulk_operations.py | 0 yatracker/tracker/categories/checklists.py | 0 yatracker/tracker/categories/comments.py | 34 +++ yatracker/tracker/categories/components.py | 0 .../tracker/categories/external_links.py | 0 yatracker/tracker/categories/import_.py | 0 yatracker/tracker/categories/issue_boards.py | 0 yatracker/tracker/categories/issue_fields.py | 0 yatracker/tracker/categories/issues.py | 247 ++++++++++++++++++ yatracker/tracker/categories/macros.py | 0 yatracker/tracker/categories/priorities.py | 22 ++ yatracker/tracker/categories/projects.py | 0 yatracker/tracker/categories/queues.py | 0 yatracker/tracker/categories/time_tracking.py | 0 yatracker/tracker/categories/users.py | 0 yatracker/types/full_issue.py | 2 + 19 files changed, 325 insertions(+), 224 deletions(-) create mode 100644 yatracker/tracker/categories/__init__.py create mode 100644 yatracker/tracker/categories/attached_files.py create mode 100644 yatracker/tracker/categories/bulk_operations.py create mode 100644 yatracker/tracker/categories/checklists.py create mode 100644 yatracker/tracker/categories/comments.py create mode 100644 yatracker/tracker/categories/components.py create mode 100644 yatracker/tracker/categories/external_links.py create mode 100644 yatracker/tracker/categories/import_.py create mode 100644 yatracker/tracker/categories/issue_boards.py create mode 100644 yatracker/tracker/categories/issue_fields.py create mode 100644 yatracker/tracker/categories/issues.py create mode 100644 yatracker/tracker/categories/macros.py create mode 100644 yatracker/tracker/categories/priorities.py create mode 100644 yatracker/tracker/categories/projects.py create mode 100644 yatracker/tracker/categories/queues.py create mode 100644 yatracker/tracker/categories/time_tracking.py create mode 100644 yatracker/tracker/categories/users.py diff --git a/yatracker/tracker/api.py b/yatracker/tracker/api.py index d7a75f9..a36b0b3 100644 --- a/yatracker/tracker/api.py +++ b/yatracker/tracker/api.py @@ -1,28 +1,23 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, Any - -from yatracker.types import ( - Comment, - FullIssue, - Priority, - Transition, - Transitions, -) from .base import BaseTracker - -if TYPE_CHECKING: - from yatracker.types import ( - Issue, - IssueType, - ) +from .categories import ( + Comments, + Issues, + Priorities, +) logger = logging.getLogger(__name__) -class YaTracker(BaseTracker): +class YaTracker( + Issues, + Comments, + Priorities, + BaseTracker, +): """Represents Yandex Tracker API client. API docs: https://cloud.yandex.com/en/docs/tracker/about-api @@ -34,211 +29,3 @@ class YaTracker(BaseTracker): For help you to recognize method names full description is attached. """ - - async def get_issue(self, issue_id: str, expand: str | None = None) -> FullIssue: - """Get issue parameters. - - Use this request to get information about an issue. - - :param issue_id: ID or key of the current issue. - :param expand: Additional fields to include in the response: - transitions — Workflow transitions between statuses. - attachments — Attachments - - :return: - """ - data = await self._client.request( - method="GET", - uri=f"/issues/{issue_id}", - params={"expand": expand} if expand else None, - ) - decoder = self._get_decoder(FullIssue) - return decoder.decode(data) - - async def edit_issue( - self, - issue_id: str, - version: str | int | None = None, - **kwargs, - ) -> FullIssue: - """Make changes to an issue. - - Use this request to make changes to an issue. - The issue is selected by its ID or key. - - Source: - https://cloud.yandex.com/en/docs/tracker/concepts/issues/patch-issue - """ - data = await self._client.request( - method="PATCH", - uri=f"/issues/{issue_id}", - params={"version": str(version)} if version else None, - payload=self.clear_payload(kwargs), - ) - decoder = self._get_decoder(FullIssue) - return decoder.decode(data) - - # ruff: noqa: ARG002 PLR0913 - async def create_issue( - self, - summary: str, - queue: str, - parent: Issue | None = None, - description: str | None = None, - sprint: dict[str, str] | None = None, - type_: IssueType | None = None, - priority: int | str | Priority | None = None, - followers: list[str] | None = None, - unique: str | None = None, - **kwargs, - ) -> FullIssue: - """Create an issue.""" - payload = self.clear_payload(locals()) - data = await self._client.request( - method="POST", - uri="/issues/", - payload=payload, - ) - decoder = self._get_decoder(FullIssue) - return decoder.decode(data) - - async def get_comments(self, issue_id: str) -> list[Comment]: - """Get the comments for an issue. - - Use this request to get a list of comments in the issue. - :param issue_id: - :return: - """ - data = await self._client.request( - method="GET", - uri=f"/issues/{issue_id}/comments", - ) - - decoder = self._get_decoder(list[Comment]) - return decoder.decode(data) - - async def post_comment(self, issue_id: str, text: str, **kwargs) -> Comment: - """Comment the issue.""" - payload = self.clear_payload(locals(), exclude=["issue_id"]) - data = await self._client.request( - method="POST", - uri=f"/issues/{issue_id}/comments/", - payload=payload, - ) - decoder = self._get_decoder(Comment) - return decoder.decode(data) - - async def count_issues( - self, - filter_: dict[str, str] | None = None, - query: str | None = None, - ) -> int: - """Get the number of issues. - - Use this request to find out how many issues meet the criteria in your request. - :return: - """ - payload: dict[str, Any] = {} - if filter_ is not None: - payload["filter"] = filter_ - if query is not None: - payload["query"] = query - - data = await self._client.request( - method="POST", - uri="/issues/_count", - payload=payload, - ) - decoder = self._get_decoder(int) - return decoder.decode(data) - - async def find_issues( - self, - filter_: dict[str, str] | None = None, - query: str | None = None, - order: str | None = None, - expand: str | None = None, - keys: str | None = None, - queue: str | None = None, - ) -> list[FullIssue]: - """Find issues. - - Use this request to get a list of issues that meet specific criteria. - If there are more than 10,000 issues in the response, use paging. - :return: - """ - payload = self.clear_payload(locals(), exclude=["expand", "order"]) - - params = {} - if order: - params["order"] = order - if expand: - params["expand"] = expand - - data = await self._client.request( - method="POST", - uri="/issues/_search", - params=params, - payload=payload, - ) - decoder = self._get_decoder(list[FullIssue]) - return decoder.decode(data) - - # ruff: noqa: FBT001 FBT002 - async def get_priorities(self, localized: bool = True) -> list[Priority]: - """Get priorities. - - Use this request to get a list of priorities for an issue. - """ - params = {"localized": str(localized).lower()} if localized else None - data = await self._client.request( - method="GET", - uri="/priorities", - params=params, - ) - - decoder = self._get_decoder(list[Priority]) - return decoder.decode(data) - - async def get_issue_links(self, issue_id: str) -> list[FullIssue]: - """Get issue links. - - Use this request to get information about links between issues. - The issue is selected by its ID or key. - """ - data = await self._client.request( - method="GET", - uri=f"/issues/{issue_id}/links", - ) - - decoder = self._get_decoder(list[FullIssue]) - return decoder.decode(data) - - async def get_transitions(self, issue_id: str) -> Transitions: - """Get transitions. - - Use this request to get a list of possible transitions for an issue. - The issue is selected by its ID or key. - """ - data = await self._client.request( - method="GET", - uri=f"/issues/{issue_id}/transitions", - ) - decoder = self._get_decoder(list[Transition]) - transitions = decoder.decode(data) - return Transitions(**{t.id: t for t in transitions}) - - async def execute_transition( - self, - transition: Transition, - **kwargs, - ) -> list[Transition]: - """Execute transition.""" - payload = self.clear_payload(kwargs) - data = await self._client.request( - method="POST", - uri=f"{transition.url}/_execute", - payload=payload, - ) - decoder = self._get_decoder(list[Transition]) - return decoder.decode(data) diff --git a/yatracker/tracker/categories/__init__.py b/yatracker/tracker/categories/__init__.py new file mode 100644 index 0000000..69eb881 --- /dev/null +++ b/yatracker/tracker/categories/__init__.py @@ -0,0 +1,9 @@ +from .comments import Comments +from .issues import Issues +from .priorities import Priorities + +__all__ = [ + "Comments", + "Issues", + "Priorities", +] diff --git a/yatracker/tracker/categories/attached_files.py b/yatracker/tracker/categories/attached_files.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/bulk_operations.py b/yatracker/tracker/categories/bulk_operations.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/checklists.py b/yatracker/tracker/categories/checklists.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/comments.py b/yatracker/tracker/categories/comments.py new file mode 100644 index 0000000..6ff5327 --- /dev/null +++ b/yatracker/tracker/categories/comments.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from yatracker.tracker.base import BaseTracker +from yatracker.types import ( + Comment, +) + + +class Comments(BaseTracker): + async def get_comments(self, issue_id: str) -> list[Comment]: + """Get the comments for an issue. + + Use this request to get a list of comments in the issue. + :param issue_id: + :return: + """ + data = await self._client.request( + method="GET", + uri=f"/issues/{issue_id}/comments", + ) + + decoder = self._get_decoder(list[Comment]) + return decoder.decode(data) + + async def post_comment(self, issue_id: str, text: str, **kwargs) -> Comment: + """Comment the issue.""" + payload = self.clear_payload(locals(), exclude=["issue_id"]) + data = await self._client.request( + method="POST", + uri=f"/issues/{issue_id}/comments/", + payload=payload, + ) + decoder = self._get_decoder(Comment) + return decoder.decode(data) diff --git a/yatracker/tracker/categories/components.py b/yatracker/tracker/categories/components.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/external_links.py b/yatracker/tracker/categories/external_links.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/import_.py b/yatracker/tracker/categories/import_.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/issue_boards.py b/yatracker/tracker/categories/issue_boards.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/issue_fields.py b/yatracker/tracker/categories/issue_fields.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/issues.py b/yatracker/tracker/categories/issues.py new file mode 100644 index 0000000..0a45a2a --- /dev/null +++ b/yatracker/tracker/categories/issues.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +from typing import Any + +from yatracker.tracker.base import BaseTracker +from yatracker.types import ( + FullIssue, + Issue, + IssueType, + Priority, + Transition, + Transitions, +) + + +class Issues(BaseTracker): + async def get_issue(self, issue_id: str, expand: str | None = None) -> FullIssue: + """Get issue parameters. + + Use this request to get information about an issue. + + :param issue_id: ID or key of the current issue. + :param expand: Additional fields to include in the response: + transitions — Workflow transitions between statuses. + attachments — Attachments + + :return: + """ + data = await self._client.request( + method="GET", + uri=f"/issues/{issue_id}", + params={"expand": expand} if expand else None, + ) + decoder = self._get_decoder(FullIssue) + return decoder.decode(data) + + async def edit_issue( + self, + issue_id: str, + version: str | int | None = None, + **kwargs, + ) -> FullIssue: + """Make changes to an issue. + + Use this request to make changes to an issue. + The issue is selected by its ID or key. + + Source: + https://cloud.yandex.com/en/docs/tracker/concepts/issues/patch-issue + """ + data = await self._client.request( + method="PATCH", + uri=f"/issues/{issue_id}", + params={"version": str(version)} if version else None, + payload=self.clear_payload(kwargs), + ) + decoder = self._get_decoder(FullIssue) + return decoder.decode(data) + + # ruff: noqa: ARG002 PLR0913 + async def create_issue( + self, + summary: str, + queue: str, + parent: Issue | None = None, + description: str | None = None, + sprint: dict[str, str] | None = None, + type_: IssueType | None = None, + priority: int | str | Priority | None = None, + followers: list[str] | None = None, + unique: str | None = None, + **kwargs, + ) -> FullIssue: + """Create an issue.""" + payload = self.clear_payload(locals()) + data = await self._client.request( + method="POST", + uri="/issues/", + payload=payload, + ) + decoder = self._get_decoder(FullIssue) + return decoder.decode(data) + + async def move_issue( + self, + issue_id: str, + queue_key: str, + *, + notify: bool = True, + notify_author: bool = False, + move_all_fields: bool = False, + initial_status: bool = False, + expand: str | None = None, + **kwargs, + ) -> FullIssue: + """Move an issue to a different queue. + + Before executing the request, make sure the user has permission + to edit the issues to be moved and is allowed to create them in + the new queue. + + Warning! + If an issue you want to move has a type and status that are + missing in the target queue, no transfer will be made. To reset + the issue status to the initial value when moving it, use the + InitialStatus parameter. + + By default, when an issue is moved, the values of its + components, versions, and projects are cleared. If the new queue + has the same values of the fields specified, use the + MoveAllFields parameter to move the components, versions, and + projects. + + If the issue has the local field values specified, they will be + reset when moving the issue to a different queue. + + You can use the request body if you need to change the + parameters of the issue being moved. The request body has the + same format as when editing issues. + + Source: + https://cloud.yandex.com/en/docs/tracker/concepts/issues/move-issue + """ + params: dict[str, Any] = {"queue": queue_key} + + if notify is False: + params["notify"] = notify + + if notify_author is True: + params["notifyAuthor"] = notify_author + + if move_all_fields is True: + params["moveAllFields"] = move_all_fields + + if initial_status is True: + params["initialStatus"] = initial_status + + if expand: + params["expand"] = expand + + data = await self._client.request( + method="POST", + uri=f"/issues/{issue_id}/_move", + params=params, + payload=self.clear_payload(kwargs), + ) + decoder = self._get_decoder(FullIssue) + return decoder.decode(data) + + async def count_issues( + self, + filter_: dict[str, str] | None = None, + query: str | None = None, + ) -> int: + """Get the number of issues. + + Use this request to find out how many issues meet the criteria in your request. + :return: + """ + payload: dict[str, Any] = {} + if filter_ is not None: + payload["filter"] = filter_ + if query is not None: + payload["query"] = query + + data = await self._client.request( + method="POST", + uri="/issues/_count", + payload=payload, + ) + decoder = self._get_decoder(int) + return decoder.decode(data) + + async def find_issues( + self, + filter_: dict[str, str] | None = None, + query: str | None = None, + order: str | None = None, + expand: str | None = None, + keys: str | None = None, + queue: str | None = None, + ) -> list[FullIssue]: + """Find issues. + + Use this request to get a list of issues that meet specific criteria. + If there are more than 10,000 issues in the response, use paging. + :return: + """ + payload = self.clear_payload(locals(), exclude=["expand", "order"]) + + params = {} + if order: + params["order"] = order + if expand: + params["expand"] = expand + + data = await self._client.request( + method="POST", + uri="/issues/_search", + params=params, + payload=payload, + ) + decoder = self._get_decoder(list[FullIssue]) + return decoder.decode(data) + + async def get_issue_links(self, issue_id: str) -> list[FullIssue]: + """Get issue links. + + Use this request to get information about links between issues. + The issue is selected by its ID or key. + """ + data = await self._client.request( + method="GET", + uri=f"/issues/{issue_id}/links", + ) + + decoder = self._get_decoder(list[FullIssue]) + return decoder.decode(data) + + async def get_transitions(self, issue_id: str) -> Transitions: + """Get transitions. + + Use this request to get a list of possible transitions for an issue. + The issue is selected by its ID or key. + """ + data = await self._client.request( + method="GET", + uri=f"/issues/{issue_id}/transitions", + ) + decoder = self._get_decoder(list[Transition]) + transitions = decoder.decode(data) + return Transitions(**{t.id: t for t in transitions}) + + async def execute_transition( + self, + transition: Transition, + **kwargs, + ) -> list[Transition]: + """Execute transition.""" + payload = self.clear_payload(kwargs) + data = await self._client.request( + method="POST", + uri=f"{transition.url}/_execute", + payload=payload, + ) + decoder = self._get_decoder(list[Transition]) + return decoder.decode(data) diff --git a/yatracker/tracker/categories/macros.py b/yatracker/tracker/categories/macros.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/priorities.py b/yatracker/tracker/categories/priorities.py new file mode 100644 index 0000000..48729df --- /dev/null +++ b/yatracker/tracker/categories/priorities.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +from yatracker.tracker.base import BaseTracker +from yatracker.types import Priority + + +class Priorities(BaseTracker): + # ruff: noqa: FBT001 FBT002 + async def get_priorities(self, localized: bool = True) -> list[Priority]: + """Get priorities. + + Use this request to get a list of priorities for an issue. + """ + params = {"localized": str(localized).lower()} if localized else None + data = await self._client.request( + method="GET", + uri="/priorities", + params=params, + ) + + decoder = self._get_decoder(list[Priority]) + return decoder.decode(data) diff --git a/yatracker/tracker/categories/projects.py b/yatracker/tracker/categories/projects.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/queues.py b/yatracker/tracker/categories/queues.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/time_tracking.py b/yatracker/tracker/categories/time_tracking.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/tracker/categories/users.py b/yatracker/tracker/categories/users.py new file mode 100644 index 0000000..e69de29 diff --git a/yatracker/types/full_issue.py b/yatracker/types/full_issue.py index 731448a..c807c9a 100644 --- a/yatracker/types/full_issue.py +++ b/yatracker/types/full_issue.py @@ -19,6 +19,7 @@ class FullIssue(Base, kw_only=True, frozen=True): id: str key: str version: int + summary: str parent: Issue | None = None description: str | None = None @@ -27,6 +28,7 @@ class FullIssue(Base, kw_only=True, frozen=True): priority: Priority followers: list[User] | None = None queue: Queue + previous_queue: Queue | None = None favorite: bool assignee: User | None = None