feat: queue automation - actions and triggers (#35)

examples/get_test_entities.py

@@ -46,6 +46,8 @@ async def main() -> None:
         await client.find_number_of_issues(issue_filter={'queue': 'TRACKER', "assignee": "empty()"})
         await client.get_history_issue_changes('TRACKER-1')
         await client.find_issues()
+        await client.get_autoactions('TRACKER')
+        await client.get_triggers('TRACKER')
     except Exception as e:
         print('Test failed')
         print(e)

ya_tracker_client/domain/client/base.py

@@ -26,12 +26,19 @@ class BaseClient(ABC):
 
     def __init__(
         self,
-        organisation_id: str | int,
+        organisation_id: str | int | None = None,
         oauth_token: str | None = None,
         iam_token: str | None = None,
         api_host: str = "https://api.tracker.yandex.net",
         api_version: str = "v2",
     ) -> None:
+        """
+        :param organisation_id: ID from admin panel at Yandex Tracker. No needed for Yandex developers.
+        :param oauth_token: OAuth token from registered application at Yandex OAuth - https://oauth.yandex.ru/
+        :param iam_token: IAM token from registered application at Yandex OAuth - https://oauth.yandex.ru/
+        :param api_host: Host of your Tracker. For Yandex developers - https://st-api.yandex-team.ru
+        :param api_version: Version of API. Currently supported only v2 version.
+        """
         self._headers: dict[str, str] = {}
 
         # Yandex 360 uses integer identifiers and Yandex Cloud prefer strings in identifiers

ya_tracker_client/domain/entities/action.py

@@ -0,0 +1,10 @@
+from pydantic import Field
+
+from ya_tracker_client.domain.entities.base import AbstractEntity
+from ya_tracker_client.domain.entities.issue_status import IssueStatus, IssueStatusKey
+
+
+class Action(AbstractEntity):
+    type: str = Field(examples=["Transition", "Update", "Event.comment-create", "Webhook", "CalculateFormula"])
+    id: int | str | None = None
+    status: IssueStatus | IssueStatusKey | None = None

ya_tracker_client/domain/entities/calendar.py

@@ -0,0 +1,5 @@
+from ya_tracker_client.domain.entities.base import AbstractEntity
+
+
+class Calendar(AbstractEntity):
+    id: int

ya_tracker_client/domain/entities/issue.py

@@ -11,7 +11,6 @@
 from ya_tracker_client.domain.entities.priority import Priority
 from ya_tracker_client.domain.entities.queue import QueueIdentifier, QueueShort
 from ya_tracker_client.domain.entities.sprint import Sprint
-from ya_tracker_client.domain.entities.status import Status
 from ya_tracker_client.domain.entities.transition import TransitionShort
 from ya_tracker_client.domain.entities.user import UserShort
 
@@ -48,8 +47,8 @@ class Issue(AbstractEntity):
     created_by: UserShort
     votes: int
     updated_at: datetime | None = None
-    status: Status
-    previous_status: Status | None = None
+    status: IssueStatus
+    previous_status: IssueStatus | None = None
     direction: str | None = None
 
     transitions: list[TransitionShort] = Field(default_factory=list)

ya_tracker_client/domain/entities/issue_status.py

@@ -6,3 +6,7 @@ class IssueStatus(AbstractEntity):
     id: str
     key: str
     display: str
+
+
+class IssueStatusKey(AbstractEntity):
+    key: str

ya_tracker_client/domain/entities/queue_autoaction.py

@@ -0,0 +1,43 @@
+from datetime import datetime
+
+from pydantic import model_validator
+
+from ya_tracker_client.domain.entities.action import Action
+from ya_tracker_client.domain.entities.base import AbstractEntity
+from ya_tracker_client.domain.entities.calendar import Calendar
+from ya_tracker_client.domain.entities.queue import QueueShort
+
+
+class Autoaction(AbstractEntity):
+    id: int | str
+    url: str
+    queue: str | QueueShort
+    name: str
+    version: int
+    active: bool
+    created: datetime
+    updated: datetime
+    filter: list[dict] | dict | None = None
+    query: str | None = None
+    actions: list[Action]
+    enable_notifications: bool
+    total_issues_processed: int
+    interval_millis: int
+    calendar: dict[str, int] | None = None
+
+
+class AutoactionCreate(AbstractEntity):
+    name: str
+    filter: list[dict] | dict | None = None
+    query: str | None = None
+    actions: list[Action]
+    active: bool | None = None
+    enable_notifications: bool | None = None
+    interval_millis: int = 3_600_000
+    calendar: Calendar | None = None
+
+    @model_validator(mode="after")
+    def filter_or_query_is_exists(self) -> "AutoactionCreate":
+        if self.filter is None and self.query is None:
+            raise ValueError("Filter or query must be not None")
+        return self

ya_tracker_client/domain/entities/trigger.py

@@ -0,0 +1,45 @@
+from pydantic import Field
+
+from ya_tracker_client.domain.entities.action import Action
+from ya_tracker_client.domain.entities.base import AbstractEntity
+from ya_tracker_client.domain.entities.queue import QueueShort
+
+
+class TriggerConditionType(AbstractEntity):
+    type: str = Field(
+        examples=[
+            "CommentNoneMatchCondition",
+            "CommentStringNotMatchCondition",
+            "CommentFullyMatchCondition",
+            "CommentAnyMatchCondition",
+            "CommentStringMatchCondition",
+            "CommentAuthorNot",
+            "CommentAuthor",
+            "CommentMessageExternal",
+            "CommentMessageInternal",
+        ],
+    )  # TODO: add validators for strings with examples (or replace it with Enum)
+
+
+class TriggerCondition(AbstractEntity):
+    type: str = Field(examples=["or", "and"])
+    conditions: list[TriggerConditionType]
+
+
+class Trigger(AbstractEntity):
+    id: str | int
+    url: str
+    queue: QueueShort | str
+    name: str
+    order: str
+    actions: list[Action]
+    conditions: list[TriggerCondition | TriggerConditionType] = Field(default_factory=list)
+    version: int
+    active: bool
+
+
+class TriggerCreate(AbstractEntity):
+    name: str
+    actions: list[Action]
+    conditions: list[TriggerCondition] | None = Field(default_factory=list)
+    active: bool | None = None

ya_tracker_client/domain/repositories/comment.py

@@ -27,7 +27,7 @@ async def add_comment(
                 summonees=summonees,
                 maillist_summonees=maillist_summonees,
             ).model_dump(exclude_none=True, by_alias=True),
-            params={"is_add_to_followers": str(is_add_to_followers).lower()}
+            params={"is_add_to_followers": str(is_add_to_followers).lower()},
         )
         return self._decode(raw_response, Comment)
 

ya_tracker_client/domain/repositories/queue.py

@@ -1,7 +1,11 @@
+from ya_tracker_client.domain.entities.action import Action
+from ya_tracker_client.domain.entities.calendar import Calendar
 from ya_tracker_client.domain.entities.issue_type_config import IssueTypeConfig
 from ya_tracker_client.domain.entities.queue import Queue, QueueCreate
+from ya_tracker_client.domain.entities.queue_autoaction import Autoaction, AutoactionCreate
 from ya_tracker_client.domain.entities.queue_field import QueueField
 from ya_tracker_client.domain.entities.queue_version import QueueVersion
+from ya_tracker_client.domain.entities.trigger import Trigger, TriggerCondition, TriggerCreate
 from ya_tracker_client.domain.repositories.base import EntityRepository
 
 
@@ -100,3 +104,115 @@ async def delete_tag_in_queue(self, queue_id: str | int, tag_name: str) -> None:
             uri=f"/queues/{queue_id}/tags/_remove",
             payload={"tag": tag_name},
         )
+
+    async def create_autoaction(
+        self,
+        queue_id: str | int,
+        name: str,
+        actions: list[Action],
+        issue_filter: list[dict] | dict | None = None,
+        query: str | None = None,
+        active: bool | None = None,
+        enable_notifications: bool | None = None,
+        interval_millis: int = 3_600_000,
+        calendar: Calendar | None = None,
+    ) -> Autoaction:
+        """
+        YC docs: https://cloud.yandex.com/en/docs/tracker/concepts/queues/create-autoaction
+        """
+        raw_response = await self._client.request(
+            method="POST",
+            uri=f"/queues/{queue_id}/autoactions",
+            payload=AutoactionCreate(
+                name=name,
+                actions=actions,
+                filter=issue_filter,
+                query=query,
+                active=active,
+                enable_notifications=enable_notifications,
+                interval_millis=interval_millis,
+                calendar=calendar,
+            ).model_dump(exclude_none=True, by_alias=True),
+        )
+        return self._decode(raw_response, Autoaction)
+
+    async def get_autoaction(
+        self,
+        queue_id: str | int,
+        autoaction_id: str | int,
+    ) -> Autoaction:
+        """
+        YC docs: https://cloud.yandex.com/en/docs/tracker/concepts/queues/get-autoaction
+        """
+        raw_response = await self._client.request(
+            method="GET",
+            uri=f"/queues/{queue_id}/autoactions/{autoaction_id}",
+        )
+        return self._decode(raw_response, Autoaction)
+
+    async def get_autoactions(
+        self,
+        queue_id: str | int,
+    ) -> list[Autoaction]:
+        # TODO: add info about this handler to YC docs
+        raw_response = await self._client.request(
+            method="GET",
+            uri=f"/queues/{queue_id}/autoactions/",
+        )
+        return self._decode(raw_response, Autoaction, plural=True)
+
+    async def delete_autoaction(
+        self,
+        queue_id: str | int,
+        autoaction_id: str | int,
+    ) -> None:
+        # TODO: add info about this handler to YC docs
+        await self._client.request(
+            method="DELETE",
+            uri=f"/queues/{queue_id}/autoactions/{autoaction_id}",
+        )
+
+    async def create_trigger(
+        self,
+        queue_id: str | int,
+        name: str,
+        actions: list[Action],
+        conditions: list[TriggerCondition] | None = None,
+        active: bool | None = None,
+    ) -> Trigger:
+        raw_response = await self._client.request(
+            method="POST",
+            uri=f"/queues/{queue_id}/triggers",
+            payload=TriggerCreate(
+                name=name,
+                actions=actions,
+                conditions=conditions,
+                active=active,
+            ).model_dump(exclude_none=True, by_alias=True),
+        )
+        return self._decode(raw_response, Trigger)
+
+    async def get_trigger(self, queue_id: str | int, trigger_id: str | int) -> Trigger:
+        """
+        YC docs: https://cloud.yandex.com/en/docs/tracker/concepts/queues/get-trigger
+        """
+        raw_response = await self._client.request(
+            method="GET",
+            uri=f"/queues/{queue_id}/triggers/{trigger_id}",
+        )
+        return self._decode(raw_response, Trigger)
+
+    async def get_triggers(self, queue_id: str | int) -> list[Trigger]:
+        # TODO: add info about this handler to YC docs
+        raw_response = await self._client.request(
+            method="GET",
+            uri=f"/queues/{queue_id}/triggers/",
+        )
+        return self._decode(raw_response, Trigger, plural=True)
+
+    async def delete_trigger(self, queue_id: str | int, trigger_id: str | int) -> None:
+        # TODO: add info about this handler to YC docs
+        await self._client.request(
+            method="DELETE",
+            uri=f"/queues/{queue_id}/triggers/{trigger_id}",
+        )