Custom fields support

examples/custom_fileds.py

@@ -0,0 +1,42 @@
+import asyncio
+
+from yatracker import YaTracker
+from yatracker.types import FullIssue
+
+# CAUTION! Don't store credentials in your code!
+ORG_ID = ...
+TOKEN = ...
+
+
+# Create your own custom Issue type:
+class HelpIssue(FullIssue, kw_only=True):
+    """Your own FullIssue type.
+
+    For example, you have some fields passed by external system.
+    One of them called 'userUsername', second - 'userId'.
+    """
+
+    user_username: str | None
+    user_id: int
+
+
+async def main() -> None:
+    """Run example."""
+    # define tracker (once)
+    tracker = YaTracker(ORG_ID, TOKEN)
+
+    # create an issue
+    issue = await tracker.create_issue(
+        summary="New Issue",
+        queue="KEY",
+        user_id=1234567890,
+        _type=HelpIssue,
+    )
+    print(issue.user_id)
+
+    # don't forget to close tracker on app shutdown (once)
+    await tracker.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

yatracker/tracker/base.py

@@ -18,6 +18,7 @@
     from .client import BaseClient
 
 T = TypeVar("T")
+B = TypeVar("B", bound=Base)
 
 logger = logging.getLogger(__name__)
 
@@ -70,18 +71,24 @@ def _decode(self, type_: type[T], data: bytes) -> T:
     def _prepare_payload(
         payload: dict[str, Any],
         exclude: Collection[str] | None = None,
+        type_: type[B] | None = None,
     ) -> dict[str, Any]:
         """Remove empty fields from payload."""
         payload = payload.copy()
         exclude = exclude or []
         kwargs = payload.pop("kwargs", None)
+
         if kwargs:
+            if type_ is not None:
+                kwargs = _replace_custom_fields(kwargs, type_)
             payload.update(kwargs)
 
         return {
             camel_case(k): _convert_value(v)
             for k, v in payload.items()
-            if k not in {"self", "cls", *exclude} and v is not None
+            if k not in {"self", "cls", *exclude}
+            and not k.startswith("_")
+            and v is not None
         }
 
     async def close(self) -> None:
@@ -133,3 +140,14 @@ def _convert_value(obj: Any) -> Any:  # noqa: ANN401
             return {k: _convert_value(v) for k, v in obj.items()}
         case _:
             return obj
+
+
+def _replace_custom_fields(kwargs: dict[str, Any], type_: type[B]) -> dict[str, Any]:
+    """Replace kwarg key with original field name."""
+    new_kwargs: dict[str, Any] = {}
+    for key, value in kwargs.items():
+        if not hasattr(type_, key):
+            continue
+        field = getattr(type_, key)
+        new_kwargs[field.name] = value
+    return new_kwargs

yatracker/tracker/categories/issues.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, TypeVar, overload
 
 from yatracker.tracker.base import BaseTracker
 from yatracker.types import (
@@ -12,9 +12,33 @@
     Transitions,
 )
 
+IssueT_co = TypeVar("IssueT_co", bound=FullIssue, covariant=True)
+
 
 class Issues(BaseTracker):
-    async def get_issue(self, issue_id: str, expand: str | None = None) -> FullIssue:
+    @overload
+    async def get_issue(
+        self,
+        issue_id: str,
+        expand: str | None = None,
+    ) -> FullIssue:
+        ...
+
+    @overload
+    async def get_issue(
+        self,
+        issue_id: str,
+        expand: str | None = None,
+        _type: type[IssueT_co] = ...,
+    ) -> IssueT_co:
+        ...
+
+    async def get_issue(
+        self,
+        issue_id: str,
+        expand: str | None = None,
+        _type: type[IssueT_co | FullIssue] = FullIssue,
+    ) -> IssueT_co | FullIssue:
         """Get issue parameters.
 
         Use this request to get information about an issue.
@@ -23,22 +47,42 @@ async def get_issue(self, issue_id: str, expand: str | None = None) -> FullIssue
         :param expand: Additional fields to include in the response:
                         transitions — Workflow transitions between statuses.
                         attachments — Attachments
-
+        :param _type: you can use your own extended FullIssue type
         :return:
         """
         data = await self._client.request(
             method="GET",
             uri=f"/issues/{issue_id}",
             params={"expand": expand} if expand else None,
         )
-        return self._decode(FullIssue, data)
+        return self._decode(_type, data)
 
+    @overload
     async def edit_issue(
         self,
         issue_id: str,
         version: str | int | None = None,
         **kwargs,
     ) -> FullIssue:
+        ...
+
+    @overload
+    async def edit_issue(
+        self,
+        issue_id: str,
+        version: str | int | None = None,
+        _type: type[IssueT_co] = ...,
+        **kwargs,
+    ) -> IssueT_co:
+        ...
+
+    async def edit_issue(
+        self,
+        issue_id: str,
+        version: str | int | None = None,
+        _type: type[IssueT_co | FullIssue] = FullIssue,
+        **kwargs,
+    ) -> IssueT_co | FullIssue:
         """Make changes to an issue.
 
         Use this request to make changes to an issue.
@@ -51,11 +95,31 @@ async def edit_issue(
             method="PATCH",
             uri=f"/issues/{issue_id}",
             params={"version": str(version)} if version else None,
-            payload=self._prepare_payload(kwargs),
+            payload=self._prepare_payload(kwargs, type_=_type),
         )
-        return self._decode(FullIssue, data)
+        return self._decode(_type, data)
 
-    # ruff: noqa: ARG002 PLR0913
+    @overload
+    async def create_issue(
+        self,
+        summary: str,
+        queue: str | int | dict,
+        *,
+        parent: Issue | str | 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,
+        assignee: list[str] | None = None,
+        unique: str | None = None,
+        attachment_ids: list[str] | None = None,
+        _type: type[IssueT_co] = ...,
+        **kwargs,
+    ) -> IssueT_co:
+        ...
+
+    @overload
     async def create_issue(
         self,
         summary: str,
@@ -72,19 +136,56 @@ async def create_issue(
         attachment_ids: list[str] | None = None,
         **kwargs,
     ) -> FullIssue:
+        ...
+
+    # ruff: noqa: ARG002 PLR0913
+    async def create_issue(
+        self,
+        summary: str,
+        queue: str | int | dict,
+        *,
+        parent: Issue | str | 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,
+        assignee: list[str] | None = None,
+        unique: str | None = None,
+        attachment_ids: list[str] | None = None,
+        _type: type[IssueT_co | FullIssue] = FullIssue,
+        **kwargs,
+    ) -> IssueT_co | FullIssue:
         """Create an issue.
 
         Source:
         https://cloud.yandex.ru/docs/tracker/concepts/issues/create-issue
         """
-        payload = self._prepare_payload(locals())
+        payload = self._prepare_payload(locals(), type_=_type)
         data = await self._client.request(
             method="POST",
             uri="/issues/",
             payload=payload,
         )
-        return self._decode(FullIssue, data)
+        return self._decode(_type, data)
+
+    @overload
+    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,
+        _type: type[IssueT_co] = ...,
+        **kwargs,
+    ) -> IssueT_co:
+        ...
 
+    @overload
     async def move_issue(
         self,
         issue_id: str,
@@ -97,6 +198,21 @@ async def move_issue(
         expand: str | None = None,
         **kwargs,
     ) -> FullIssue:
+        ...
+
+    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,
+        _type: type[IssueT_co | FullIssue] = FullIssue,
+        **kwargs,
+    ) -> IssueT_co | FullIssue:
         """Move an issue to a different queue.
 
         Before executing the request, make sure the user has permission
@@ -146,9 +262,9 @@ async def move_issue(
             method="POST",
             uri=f"/issues/{issue_id}/_move",
             params=params,
-            payload=self._prepare_payload(kwargs),
+            payload=self._prepare_payload(kwargs, type_=_type),
         )
-        return self._decode(FullIssue, data)
+        return self._decode(_type, data)
 
     async def count_issues(
         self,
@@ -173,6 +289,7 @@ async def count_issues(
         )
         return self._decode(int, data)
 
+    @overload
     async def find_issues(
         self,
         filter_: dict[str, str] | None = None,
@@ -182,13 +299,42 @@ async def find_issues(
         keys: str | None = None,
         queue: str | None = None,
     ) -> list[FullIssue]:
+        ...
+
+    @overload
+    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,
+        _type: type[IssueT_co] = ...,
+    ) -> list[IssueT_co]:
+        ...
+
+    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,
+        _type: type[IssueT_co | FullIssue] = FullIssue,
+    ) -> list[IssueT_co] | 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._prepare_payload(locals(), exclude=["expand", "order"])
+        payload = self._prepare_payload(
+            locals(),
+            exclude=["expand", "order"],
+            type_=_type,
+        )
 
         params = {}
         if order:
@@ -202,9 +348,28 @@ async def find_issues(
             params=params,
             payload=payload,
         )
-        return self._decode(list[FullIssue], data)
+        return self._decode(list[_type], data)  # type: ignore[valid-type]
+
+    @overload
+    async def get_issue_links(
+        self,
+        issue_id: str,
+    ) -> list[FullIssue]:
+        ...
 
-    async def get_issue_links(self, issue_id: str) -> list[FullIssue]:
+    @overload
+    async def get_issue_links(
+        self,
+        issue_id: str,
+        _type: type[IssueT_co | FullIssue] = ...,
+    ) -> list[IssueT_co]:
+        ...
+
+    async def get_issue_links(
+        self,
+        issue_id: str,
+        _type: type[IssueT_co | FullIssue] = FullIssue,
+    ) -> list[IssueT_co] | list[FullIssue]:
         """Get issue links.
 
         Use this request to get information about links between issues.
@@ -214,7 +379,7 @@ async def get_issue_links(self, issue_id: str) -> list[FullIssue]:
             method="GET",
             uri=f"/issues/{issue_id}/links",
         )
-        return self._decode(list[FullIssue], data)
+        return self._decode(list[_type], data)  # type: ignore[valid-type]
 
     async def get_transitions(self, issue_id: str) -> Transitions:
         """Get transitions.