[Types]: adding mypy type annotations to a lot of the repo

Author: jpetrucciani

.prospector.yaml

@@ -6,6 +6,8 @@ test-warnings: false
 ignore-patterns:
   - (^|/)\..+
   - .*\.html
+  - docs/.*
+  - tmp.py
 
 pylint:
   disable:

Makefile

@@ -1,7 +1,7 @@
 repo = pyclickup
 base_command = pytest
 coverage = --cov-config setup.cfg --cov=$(repo)
-with_report = --cov-report html
+html_report = --cov-report html
 term_report = --cov-report term
 xml_report = --cov-report xml
 reports = $(html_report) $(term_report) $(xml_report)

README.rst

@@ -49,7 +49,7 @@ Basic Usage
    from pyclickup import ClickUp
 
 
-   clickup = ClickUp('$ACCESS_TOKEN')
+   clickup = ClickUp("$ACCESS_TOKEN")
 
    main_team = clickup.teams[0]
    main_space = main_team.spaces[0]

pyclickup/globals.py

@@ -3,7 +3,7 @@
 """
 
 
-__version__ = "0.0.7"
+__version__ = "0.0.8"
 
 
 LIBRARY = "pyclickup"

pyclickup/models/__init__.py

@@ -4,12 +4,13 @@
 import json
 from pyclickup.globals import DEFAULT_STATUSES, LIBRARY
 from pyclickup.utils.text import snakeify, ts_to_datetime, datetime_to_ts
+from typing import Any
 
 
 class BaseModel:
     """basic model that just parses camelCase json to snake_case keys"""
 
-    def __init__(self, data, client=None, **kwargs):
+    def __init__(self, data: dict, client: Any = None, **kwargs: Any) -> None:
         """constructor"""
         self._data = {**data, **kwargs}
         self._json = self._jsond(data)
@@ -18,19 +19,19 @@ def __init__(self, data, client=None, **kwargs):
         for key in self._data:
             setattr(self, snakeify(key), self._data[key])
 
-    def _jsond(self, json_data):
+    def _jsond(self, json_data: dict) -> str:
         """json dumps"""
         return json.dumps(json_data)
 
-    def _jsonl(self, dictionary):
+    def _jsonl(self, dictionary: str) -> dict:
         """json loads"""
         return json.loads(dictionary)
 
 
 class User(BaseModel):
     """user object"""
 
-    def __init__(self, data, **kwargs):
+    def __init__(self, data: dict, **kwargs: Any) -> None:
         """override"""
         if "user" in data.keys():
             data = data["user"]

pyclickup/models/client.py

@@ -3,15 +3,13 @@
 """
 import requests
 import urllib.parse
+from datetime import datetime
+from requests.models import Response
 from pyclickup.globals import __version__, API_URL, LIBRARY, TEST_TOKEN, TEST_API_URL
 from pyclickup.models import User, Task, Team
 from pyclickup.models.error import RateLimited
 from pyclickup.utils.text import datetime_to_ts, filter_locals
-
-
-def test_client():
-    """returns a test client"""
-    return ClickUp(TEST_TOKEN, api_url=TEST_API_URL, debug=True)
+from typing import Any, Dict, List, Optional, Union  # noqa
 
 
 class ClickUp:
@@ -26,7 +24,13 @@ class ClickUp:
         "assignees",
     ]
 
-    def __init__(self, token, api_url=API_URL, cache=True, debug=False):
+    def __init__(
+        self,
+        token: str,
+        api_url: str = API_URL,
+        cache: bool = True,
+        debug: bool = False,
+    ) -> None:
         """creates a new client"""
         if not token:
             raise Exception("no token specified!")
@@ -37,11 +41,11 @@ def __init__(self, token, api_url=API_URL, cache=True, debug=False):
         self.debug = debug
 
         # cache
-        self._user = None
-        self._teams = None
+        self._user = None  # type: Optional[User]
+        self._teams = None  # type: Optional[List[Team]]
 
     @property
-    def headers(self):
+    def headers(self) -> dict:
         """forms the headers required for the API calls"""
         return {
             "Accept": "application/json",
@@ -51,30 +55,36 @@ def headers(self):
         }
 
     @property
-    def user(self):
+    def user(self) -> User:
         """get the user associated with this token"""
         if not self._user or not self.cache:
             self._user = User(self.get("user"), client=self)
         return self._user
 
     @property
-    def teams(self):
+    def teams(self) -> List[Team]:
         """get authorized teams"""
         if not self._teams or not self.cache:
-            self._teams = [Team(x, client=self) for x in self.get("team")["teams"]]
+            teams_data = self.get("team")
+            if not isinstance(teams_data, dict):
+                raise Exception("invalid response while looking up teams")
+            self._teams = [Team(x, client=self) for x in teams_data["teams"]]
         return self._teams
 
-    def get_team_by_id(self, team_id):
+    def get_team_by_id(self, team_id: str) -> Team:
         """given an team_id, return the team if it exists"""
-        return Team(self.get("team/{}".format(team_id))["team"], client=self)
+        team_data = self.get("team/{}".format(team_id))
+        if not isinstance(team_data, dict):
+            raise Exception("no team found")
+        return Team(team_data["team"], client=self)
 
-    def _log(self, *args):
+    def _log(self, *args: Any) -> None:
         """logging method"""
         if not self.debug:
             return
         print(*args)
 
-    def _req(self, path, method="get", **kwargs):
+    def _req(self, path: str, method: str = "get", **kwargs: Any) -> Response:
         """requests wrapper"""
         full_path = urllib.parse.urljoin(self.api_url, path)
         self._log("[{}]: {}".format(method.upper(), full_path))
@@ -83,39 +93,48 @@ def _req(self, path, method="get", **kwargs):
             raise RateLimited()
         return request
 
-    def get(self, path, **kwargs):
+    def get(
+        self, path: str, raw: bool = False, **kwargs: Any
+    ) -> Union[list, dict, Response]:
         """makes a get request to the API"""
-        return self._req(path, **kwargs).json()
+        request = self._req(path, **kwargs)
+        return request if raw else request.json()
 
-    def post(self, path, **kwargs):
+    def post(
+        self, path: str, raw: bool = False, **kwargs: Any
+    ) -> Union[list, dict, Response]:
         """makes a post request to the API"""
-        return self._req(path, method="post", **kwargs).json()
+        request = self._req(path, method="post", **kwargs)
+        return request if raw else request.json()
 
-    def put(self, path, **kwargs):
+    def put(
+        self, path: str, raw: bool = False, **kwargs: Any
+    ) -> Union[list, dict, Response]:
         """makes a put request to the API"""
-        return self._req(path, method="put", **kwargs).json()
+        request = self._req(path, method="put", **kwargs)
+        return request if raw else request.json()
 
     def _get_tasks(
         self,
-        team_id,
-        page=None,  # integer - it appears to fetch 100 at a time
-        order_by=None,  # string, [id, created, updated, due_date]
-        reverse=None,  # bool
-        subtasks=None,  # bool
-        space_ids=None,  # List
-        project_ids=None,  # List
-        list_ids=None,  # List
-        statuses=None,  # List
-        include_closed=False,  # bool
-        assignees=None,  # List
-        due_date_gt=None,  # integer, posix time
-        due_date_lt=None,  # integer, posix time
-        date_created_gt=None,  # integer, posix time
-        date_created_lt=None,  # integer, posix time
-        date_updated_gt=None,  # integer, posix time
-        date_updated_lt=None,  # integer, posix time
-        **kwargs
-    ):
+        team_id: str,
+        page: int = None,  # integer - it appears to fetch 100 at a time
+        order_by: str = None,  # string, [id, created, updated, due_date]
+        reverse: bool = None,  # bool
+        subtasks: bool = None,  # bool
+        space_ids: list = None,  # List
+        project_ids: list = None,  # List
+        list_ids: list = None,  # List
+        statuses: list = None,  # List
+        include_closed: bool = False,  # bool
+        assignees: list = None,  # List
+        due_date_gt: int = None,  # integer, posix time
+        due_date_lt: int = None,  # integer, posix time
+        date_created_gt: int = None,  # integer, posix time
+        date_created_lt: int = None,  # integer, posix time
+        date_updated_gt: int = None,  # integer, posix time
+        date_updated_lt: int = None,  # integer, posix time
+        **kwargs: Any
+    ) -> List[Task]:
         """fetches the tasks according to the given options"""
         params = filter_locals(locals(), extras=["team_id"])
 
@@ -132,31 +151,40 @@ def _get_tasks(
             for x in params
         ]
         path = "team/{}/task?{}".format(team_id, "&".join(options))
-        return [Task(x, client=self) for x in self.get(path)["tasks"]]
-
-    def _get_all_tasks(self, team_id, **kwargs):
+        task_list = self.get(path)
+        if not isinstance(task_list, dict):
+            return []
+        return [Task(x, client=self) for x in task_list["tasks"]]
+
+    def _get_all_tasks(
+        self, team_id: str, page_limit: int = -1, **kwargs: Any
+    ) -> List[Task]:
         """get all tasks wrapper"""
-        tasks = []
+        tasks = []  # type: List[Task]
         page_count = 0
         task_page = self._get_tasks(team_id, page=page_count, **kwargs)
-        while task_page:
+        while task_page and (page_limit == -1 or page_count < page_limit):
             tasks += task_page
             page_count += 1
             task_page = self._get_tasks(team_id, page=page_count, **kwargs)
         return tasks
 
     def _create_task(
         self,
-        list_id,
-        name,  # string
-        content,  # string
-        status,  # string
-        assignees=None,  # array
-        priority=None,  # integer
-        due_date=None,  # integer posix time, or python datetime
-    ):
+        list_id: str,
+        name: str,  # string
+        content: str,  # string
+        status: str,  # string
+        assignees: list = None,  # array
+        priority: int = None,  # integer
+        due_date: Union[int, datetime] = None,  # integer posix time, or python datetime
+    ) -> Any:
         """creates a task in the specified list"""
-        data = {"name": name, "content": content, "status": status}
+        data = {
+            "name": name,
+            "content": content,
+            "status": status,
+        }  # type: Dict[str, Any]
         if assignees:
             data["assignees"] = assignees
         if priority:
@@ -166,3 +194,8 @@ def _create_task(
                 due_date if isinstance(due_date, int) else datetime_to_ts(due_date)
             )
         return self.post("list/{}/task".format(list_id), data=data)
+
+
+def test_client() -> ClickUp:
+    """returns a test client"""
+    return ClickUp(TEST_TOKEN, api_url=TEST_API_URL, debug=True)

pyclickup/models/error.py

@@ -1,12 +1,13 @@
 """
 error models for pyclickup
 """
+from typing import Any
 
 
 class PyClickUpException(Exception):
     """base pyclickup exception class"""
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         extra = ""
         if args:
             extra = '\n| extra info: "{extra}"'.format(extra=args[0])

pyclickup/test/conftest.py

@@ -3,15 +3,16 @@
 """
 import pytest
 from pyclickup.test.helpers import dbg
+from typing import Any
 
 
 @pytest.fixture(scope="session", autouse=True)
-def before_all(request):
+def before_all(request: Any) -> None:
     """test setup"""
     dbg("[+] begin pyclickup tests")
     request.addfinalizer(after_all)
 
 
-def after_all():
+def after_all() -> None:
     """tear down"""
     dbg("[+] end pyclickup tests")

pyclickup/test/helpers.py

@@ -9,12 +9,7 @@
 init()
 
 
-def manual_raise():
-    """manually raise an exception"""
-    raise SystemExit(1)
-
-
-def dbg(text):
+def dbg(text: str) -> None:
     """debug printer for tests"""
     if isinstance(text, dict):
         text = json.dumps(text, sort_keys=True, indent=2)