Add methods to get completed tasks

tests/conftest.py

@@ -10,6 +10,7 @@
     DEFAULT_COLLABORATORS_RESPONSE,
     DEFAULT_COMMENT_RESPONSE,
     DEFAULT_COMMENTS_RESPONSE,
+    DEFAULT_COMPLETED_TASKS_RESPONSE,
     DEFAULT_LABEL_RESPONSE,
     DEFAULT_LABELS_RESPONSE,
     DEFAULT_PROJECT_RESPONSE,
@@ -20,6 +21,7 @@
     DEFAULT_TASK_RESPONSE,
     DEFAULT_TASKS_RESPONSE,
     DEFAULT_TOKEN,
+    PaginatedItems,
     PaginatedResults,
 )
 from todoist_api_python.api import TodoistAPI
@@ -87,6 +89,19 @@ def default_tasks_list() -> list[list[Task]]:
     ]
 
 
+@pytest.fixture
+def default_completed_tasks_response() -> list[PaginatedItems]:
+    return DEFAULT_COMPLETED_TASKS_RESPONSE
+
+
+@pytest.fixture
+def default_completed_tasks_list() -> list[list[Task]]:
+    return [
+        [Task.from_dict(result) for result in response["items"]]
+        for response in DEFAULT_COMPLETED_TASKS_RESPONSE
+    ]
+
+
 @pytest.fixture
 def default_project_response() -> dict[str, Any]:
     return DEFAULT_PROJECT_RESPONSE

tests/data/test_defaults.py

@@ -8,6 +8,11 @@ class PaginatedResults(TypedDict):
     next_cursor: str | None
 
 
+class PaginatedItems(TypedDict):
+    items: list[dict[str, Any]]
+    next_cursor: str | None
+
+
 DEFAULT_API_URL = "https://api.todoist.com/api/v1"
 DEFAULT_OAUTH_URL = "https://todoist.com/oauth"
 
@@ -114,6 +119,28 @@ class PaginatedResults(TypedDict):
 DEFAULT_TASK_META_RESPONSE = dict(DEFAULT_TASK_RESPONSE)
 DEFAULT_TASK_META_RESPONSE["meta"] = DEFAULT_META_RESPONSE
 
+DEFAULT_COMPLETED_TASK_RESPONSE = dict(DEFAULT_TASK_RESPONSE)
+DEFAULT_COMPLETED_TASK_RESPONSE["completed_at"] = "2024-02-13T10:00:00.000000Z"
+
+DEFAULT_COMPLETED_TASK_RESPONSE_2 = dict(DEFAULT_COMPLETED_TASK_RESPONSE)
+DEFAULT_COMPLETED_TASK_RESPONSE_2["id"] = "6X7rfFVPjhvv84XG"
+
+DEFAULT_COMPLETED_TASK_RESPONSE_3 = dict(DEFAULT_COMPLETED_TASK_RESPONSE)
+DEFAULT_COMPLETED_TASK_RESPONSE_3["id"] = "6X7rfEVP8hvv25ZQ"
+
+DEFAULT_COMPLETED_TASKS_RESPONSE: list[PaginatedItems] = [
+    {
+        "items": [
+            DEFAULT_COMPLETED_TASK_RESPONSE,
+            DEFAULT_COMPLETED_TASK_RESPONSE_2,
+        ],
+        "next_cursor": "next",
+    },
+    {
+        "items": [DEFAULT_COMPLETED_TASK_RESPONSE_3],
+        "next_cursor": None,
+    },
+]
 
 DEFAULT_COLLABORATOR_RESPONSE = {
     "id": "6X7rM8997g3RQmvh",

tests/test_api_completed_tasks.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
+
+import pytest
+import responses
+
+from tests.data.test_defaults import DEFAULT_API_URL, PaginatedItems
+from tests.utils.test_utils import auth_matcher, enumerate_async, param_matcher
+from todoist_api_python._core.utils import format_datetime
+
+if TYPE_CHECKING:
+    from todoist_api_python.api import TodoistAPI
+    from todoist_api_python.api_async import TodoistAPIAsync
+    from todoist_api_python.models import Task
+
+
+@pytest.mark.asyncio
+async def test_get_completed_tasks_by_due_date(
+    todoist_api: TodoistAPI,
+    todoist_api_async: TodoistAPIAsync,
+    requests_mock: responses.RequestsMock,
+    default_completed_tasks_response: list[PaginatedItems],
+    default_completed_tasks_list: list[list[Task]],
+) -> None:
+    since = datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC)
+    until = datetime(2024, 2, 1, 0, 0, 0, tzinfo=UTC)
+    project_id = "6X7rM8997g3RQmvh"
+    filter_query = "p1"
+
+    params = {
+        "since": format_datetime(since),
+        "until": format_datetime(until),
+        "project_id": project_id,
+        "filter_query": filter_query,
+    }
+
+    endpoint = f"{DEFAULT_API_URL}/tasks/completed/by_due_date"
+
+    cursor: str | None = None
+    for page in default_completed_tasks_response:
+        requests_mock.add(
+            method=responses.GET,
+            url=endpoint,
+            json=page,
+            status=200,
+            match=[auth_matcher(), param_matcher(params, cursor)],
+        )
+        cursor = page["next_cursor"]
+
+    count = 0
+
+    tasks_iter = todoist_api.get_completed_tasks_by_due_date(
+        since=since,
+        until=until,
+        project_id=project_id,
+        filter_query=filter_query,
+    )
+
+    for i, tasks in enumerate(tasks_iter):
+        assert len(requests_mock.calls) == count + 1
+        assert tasks == default_completed_tasks_list[i]
+        count += 1
+
+    tasks_async_iter = await todoist_api_async.get_completed_tasks_by_due_date(
+        since=since,
+        until=until,
+        project_id=project_id,
+        filter_query=filter_query,
+    )
+
+    async for i, tasks in enumerate_async(tasks_async_iter):
+        assert len(requests_mock.calls) == count + 1
+        assert tasks == default_completed_tasks_list[i]
+        count += 1
+
+
+@pytest.mark.asyncio
+async def test_get_completed_tasks_by_completion_date(
+    todoist_api: TodoistAPI,
+    todoist_api_async: TodoistAPIAsync,
+    requests_mock: responses.RequestsMock,
+    default_completed_tasks_response: list[PaginatedItems],
+    default_completed_tasks_list: list[list[Task]],
+) -> None:
+    since = datetime(2024, 3, 1, 0, 0, 0)  # noqa: DTZ001
+    until = datetime(2024, 4, 1, 0, 0, 0)  # noqa: DTZ001
+    workspace_id = "123"
+    filter_query = "@label"
+
+    params: dict[str, Any] = {
+        "since": format_datetime(since),
+        "until": format_datetime(until),
+        "workspace_id": workspace_id,
+        "filter_query": filter_query,
+    }
+
+    endpoint = f"{DEFAULT_API_URL}/tasks/completed/by_completion_date"
+
+    cursor: str | None = None
+    for page in default_completed_tasks_response:
+        requests_mock.add(
+            method=responses.GET,
+            url=endpoint,
+            json=page,
+            status=200,
+            match=[auth_matcher(), param_matcher(params, cursor)],
+        )
+        cursor = page["next_cursor"]
+
+    count = 0
+
+    tasks_iter = todoist_api.get_completed_tasks_by_completion_date(
+        since=since,
+        until=until,
+        workspace_id=workspace_id,
+        filter_query=filter_query,
+    )
+
+    for i, tasks in enumerate(tasks_iter):
+        assert len(requests_mock.calls) == count + 1
+        assert tasks == default_completed_tasks_list[i]
+        count += 1
+
+    tasks_async_iter = await todoist_api_async.get_completed_tasks_by_completion_date(
+        since=since,
+        until=until,
+        workspace_id=workspace_id,
+        filter_query=filter_query,
+    )
+
+    async for i, tasks in enumerate_async(tasks_async_iter):
+        assert len(requests_mock.calls) == count + 1
+        assert tasks == default_completed_tasks_list[i]
+        count += 1

todoist_api_python/_core/endpoints.py

@@ -14,6 +14,9 @@
 TASKS_PATH = "tasks"
 TASKS_FILTER_PATH = "tasks/filter"
 TASKS_QUICK_ADD_PATH = "tasks/quick"
+TASKS_COMPLETED_PATH = "tasks/completed"
+TASKS_COMPLETED_BY_DUE_DATE_PATH = f"{TASKS_COMPLETED_PATH}/by_due_date"
+TASKS_COMPLETED_BY_COMPLETION_DATE_PATH = f"{TASKS_COMPLETED_PATH}/by_completion_date"
 PROJECTS_PATH = "projects"
 COLLABORATORS_PATH = "collaborators"
 SECTIONS_PATH = "sections"

todoist_api_python/api.py

@@ -16,6 +16,8 @@
     SHARED_LABELS_PATH,
     SHARED_LABELS_REMOVE_PATH,
     SHARED_LABELS_RENAME_PATH,
+    TASKS_COMPLETED_BY_COMPLETION_DATE_PATH,
+    TASKS_COMPLETED_BY_DUE_DATE_PATH,
     TASKS_FILTER_PATH,
     TASKS_PATH,
     TASKS_QUICK_ADD_PATH,
@@ -477,6 +479,116 @@ def delete_task(self, task_id: str) -> bool:
         endpoint = get_api_url(f"{TASKS_PATH}/{task_id}")
         return delete(self._session, endpoint, self._token)
 
+    def get_completed_tasks_by_due_date(
+        self,
+        *,
+        since: datetime,
+        until: datetime,
+        workspace_id: str | None = None,
+        project_id: str | None = None,
+        section_id: str | None = None,
+        parent_id: str | None = None,
+        filter_query: str | None = None,
+        filter_lang: str | None = None,
+        limit: Annotated[int, Ge(1), Le(200)] | None = None,
+    ) -> Iterator[list[Task]]:
+        """
+        Get an iterable of lists of completed tasks within a due date range.
+
+        Retrieves tasks completed within a specific due date range (up to 6 weeks).
+        Supports filtering by workspace, project, section, parent task, or a query.
+
+        The response is an iterable of lists of completed tasks. Be aware that each
+        iteration fires off a network request to the Todoist API, and may result in
+        rate limiting or other API restrictions.
+
+        :param since: Start of the date range (inclusive).
+        :param until: End of the date range (inclusive).
+        :param workspace_id: Filter by workspace ID.
+        :param project_id: Filter by project ID.
+        :param section_id: Filter by section ID.
+        :param parent_id: Filter by parent task ID.
+        :param filter_query: Filter by a query string.
+        :param filter_lang: Language for the filter query (e.g., 'en').
+        :param limit: Maximum number of tasks per page (default 50).
+        :return: An iterable of lists of completed tasks.
+        :raises requests.exceptions.HTTPError: If the API request fails.
+        :raises TypeError: If the API response structure is unexpected.
+        """
+        endpoint = get_api_url(TASKS_COMPLETED_BY_DUE_DATE_PATH)
+
+        params: dict[str, Any] = {
+            "since": format_datetime(since),
+            "until": format_datetime(until),
+        }
+        if workspace_id is not None:
+            params["workspace_id"] = workspace_id
+        if project_id is not None:
+            params["project_id"] = project_id
+        if section_id is not None:
+            params["section_id"] = section_id
+        if parent_id is not None:
+            params["parent_id"] = parent_id
+        if filter_query is not None:
+            params["filter_query"] = filter_query
+        if filter_lang is not None:
+            params["filter_lang"] = filter_lang
+        if limit is not None:
+            params["limit"] = limit
+
+        return ResultsPaginator(
+            self._session, endpoint, "items", Task.from_dict, self._token, params
+        )
+
+    def get_completed_tasks_by_completion_date(
+        self,
+        *,
+        since: datetime,
+        until: datetime,
+        workspace_id: str | None = None,
+        filter_query: str | None = None,
+        filter_lang: str | None = None,
+        limit: Annotated[int, Ge(1), Le(200)] | None = None,
+    ) -> Iterator[list[Task]]:
+        """
+        Get an iterable of lists of completed tasks within a date range.
+
+        Retrieves tasks completed within a specific date range (up to 3 months).
+        Supports filtering by workspace or a filter query.
+
+        The response is an iterable of lists of completed tasks. Be aware that each
+        iteration fires off a network request to the Todoist API, and may result in
+        rate limiting or other API restrictions.
+
+        :param since: Start of the date range (inclusive).
+        :param until: End of the date range (inclusive).
+        :param workspace_id: Filter by workspace ID.
+        :param filter_query: Filter by a query string.
+        :param filter_lang: Language for the filter query (e.g., 'en').
+        :param limit: Maximum number of tasks per page (default 50).
+        :return: An iterable of lists of completed tasks.
+        :raises requests.exceptions.HTTPError: If the API request fails.
+        :raises TypeError: If the API response structure is unexpected.
+        """
+        endpoint = get_api_url(TASKS_COMPLETED_BY_COMPLETION_DATE_PATH)
+
+        params: dict[str, Any] = {
+            "since": format_datetime(since),
+            "until": format_datetime(until),
+        }
+        if workspace_id is not None:
+            params["workspace_id"] = workspace_id
+        if filter_query is not None:
+            params["filter_query"] = filter_query
+        if filter_lang is not None:
+            params["filter_lang"] = filter_lang
+        if limit is not None:
+            params["limit"] = limit
+
+        return ResultsPaginator(
+            self._session, endpoint, "items", Task.from_dict, self._token, params
+        )
+
     def get_project(self, project_id: str) -> Project:
         """
         Get a project by its ID.