python-gitlab · nejch · Nov 23, 2021 · Nov 22, 2021 · Nov 17, 2021 · Nov 21, 2021
diff --git a/gitlab/base.py b/gitlab/base.py
@@ -150,6 +150,10 @@ def _create_managers(self) -> None:
         # annotations. If an attribute is annotated as being a *Manager type
         # then we create the manager and assign it to the attribute.
         for attr, annotation in sorted(self.__annotations__.items()):
+            # We ignore creating a manager for the 'manager' attribute as that
+            # is done in the self.__init__() method
+            if attr in ("manager",):
+                continue
             if not isinstance(annotation, (type, str)):
                 continue
             if isinstance(annotation, type):

diff --git a/gitlab/v4/objects/epics.py b/gitlab/v4/objects/epics.py
@@ -1,3 +1,5 @@
+from typing import Any, cast, Dict, Optional, TYPE_CHECKING, Union
+
 from gitlab import exceptions as exc
 from gitlab import types
 from gitlab.base import RequiredOptional, RESTManager, RESTObject
@@ -42,11 +44,17 @@ class GroupEpicManager(CRUDMixin, RESTManager):
     )
     _types = {"labels": types.ListAttribute}
 
+    def get(self, id: Union[str, int], lazy: bool = False, **kwargs: Any) -> GroupEpic:
+        return cast(GroupEpic, super().get(id=id, lazy=lazy, **kwargs))
+
 
 class GroupEpicIssue(ObjectDeleteMixin, SaveMixin, RESTObject):
     _id_attr = "epic_issue_id"
+    # Define type for 'manager' here So mypy won't complain about
+    # 'self.manager.update()' call in the 'save' method.
+    manager: "GroupEpicIssueManager"
 
-    def save(self, **kwargs):
+    def save(self, **kwargs: Any) -> None:
         """Save the changes made to the object to the server.
 
         The object is updated to match what the server returns.
@@ -78,7 +86,9 @@ class GroupEpicIssueManager(
     _update_attrs = RequiredOptional(optional=("move_before_id", "move_after_id"))
 
     @exc.on_http_error(exc.GitlabCreateError)
-    def create(self, data, **kwargs):
+    def create(
+        self, data: Optional[Dict[str, Any]] = None, **kwargs: Any
+    ) -> GroupEpicIssue:
         """Create a new object.
 
         Args:
@@ -94,9 +104,13 @@ def create(self, data, **kwargs):
             RESTObject: A new instance of the manage object class build with
                         the data sent by the server
         """
+        if TYPE_CHECKING:
+            assert data is not None
         CreateMixin._check_missing_create_attrs(self, data)
         path = f"{self.path}/{data.pop('issue_id')}"
         server_data = self.gitlab.http_post(path, **kwargs)
+        if TYPE_CHECKING:
+            assert isinstance(server_data, dict)
         # The epic_issue_id attribute doesn't exist when creating the resource,
         # but is used everywhere elese. Let's create it to be consistent client
         # side

diff --git a/gitlab/v4/objects/files.py b/gitlab/v4/objects/files.py
@@ -1,4 +1,7 @@
 import base64
+from typing import Any, Callable, cast, Dict, List, Optional, TYPE_CHECKING
+
+import requests
 
 from gitlab import cli
 from gitlab import exceptions as exc
@@ -22,6 +25,8 @@
 class ProjectFile(SaveMixin, ObjectDeleteMixin, RESTObject):
     _id_attr = "file_path"
     _short_print_attr = "file_path"
+    file_path: str
+    manager: "ProjectFileManager"
 
     def decode(self) -> bytes:
         """Returns the decoded content of the file.
@@ -31,7 +36,11 @@ def decode(self) -> bytes:
         """
         return base64.b64decode(self.content)
 
-    def save(self, branch, commit_message, **kwargs):
+    # NOTE(jlvillal): Signature doesn't match SaveMixin.save() so ignore
+    # type error
+    def save(  # type: ignore
+        self, branch: str, commit_message: str, **kwargs: Any
+    ) -> None:
         """Save the changes made to the file to the server.
 
         The object is updated to match what the server returns.
@@ -50,7 +59,12 @@ def save(self, branch, commit_message, **kwargs):
         self.file_path = self.file_path.replace("/", "%2F")
         super(ProjectFile, self).save(**kwargs)
 
-    def delete(self, branch, commit_message, **kwargs):
+    @exc.on_http_error(exc.GitlabDeleteError)
+    # NOTE(jlvillal): Signature doesn't match DeleteMixin.delete() so ignore
+    # type error
+    def delete(  # type: ignore
+        self, branch: str, commit_message: str, **kwargs: Any
+    ) -> None:
         """Delete the file from the server.
 
         Args:
@@ -80,7 +94,11 @@ class ProjectFileManager(GetMixin, CreateMixin, UpdateMixin, DeleteMixin, RESTMa
     )
 
     @cli.register_custom_action("ProjectFileManager", ("file_path", "ref"))
-    def get(self, file_path, ref, **kwargs):
+    # NOTE(jlvillal): Signature doesn't match UpdateMixin.update() so ignore
+    # type error
+    def get(  # type: ignore
+        self, file_path: str, ref: str, **kwargs: Any
+    ) -> ProjectFile:
         """Retrieve a single file.
 
         Args:
@@ -95,15 +113,17 @@ def get(self, file_path, ref, **kwargs):
         Returns:
             object: The generated RESTObject
         """
-        return GetMixin.get(self, file_path, ref=ref, **kwargs)
+        return cast(ProjectFile, GetMixin.get(self, file_path, ref=ref, **kwargs))
 
     @cli.register_custom_action(
         "ProjectFileManager",
         ("file_path", "branch", "content", "commit_message"),
         ("encoding", "author_email", "author_name"),
     )
     @exc.on_http_error(exc.GitlabCreateError)
-    def create(self, data, **kwargs):
+    def create(
+        self, data: Optional[Dict[str, Any]] = None, **kwargs: Any
+    ) -> ProjectFile:
         """Create a new object.
 
         Args:
@@ -120,15 +140,23 @@ def create(self, data, **kwargs):
             GitlabCreateError: If the server cannot perform the request
         """
 
+        if TYPE_CHECKING:
+            assert data is not None
         self._check_missing_create_attrs(data)
         new_data = data.copy()
         file_path = new_data.pop("file_path").replace("/", "%2F")
         path = f"{self.path}/{file_path}"
         server_data = self.gitlab.http_post(path, post_data=new_data, **kwargs)
+        if TYPE_CHECKING:
+            assert isinstance(server_data, dict)
         return self._obj_cls(self, server_data)
 
     @exc.on_http_error(exc.GitlabUpdateError)
-    def update(self, file_path, new_data=None, **kwargs):
+    # NOTE(jlvillal): Signature doesn't match UpdateMixin.update() so ignore
+    # type error
+    def update(  # type: ignore
+        self, file_path: str, new_data: Optional[Dict[str, Any]] = None, **kwargs: Any
+    ) -> Dict[str, Any]:
         """Update an object on the server.
 
         Args:
@@ -149,13 +177,20 @@ def update(self, file_path, new_data=None, **kwargs):
         data["file_path"] = file_path
         path = f"{self.path}/{file_path}"
         self._check_missing_update_attrs(data)
-        return self.gitlab.http_put(path, post_data=data, **kwargs)
+        result = self.gitlab.http_put(path, post_data=data, **kwargs)
+        if TYPE_CHECKING:
+            assert isinstance(result, dict)
+        return result
 
     @cli.register_custom_action(
         "ProjectFileManager", ("file_path", "branch", "commit_message")
     )
     @exc.on_http_error(exc.GitlabDeleteError)
-    def delete(self, file_path, branch, commit_message, **kwargs):
+    # NOTE(jlvillal): Signature doesn't match DeleteMixin.delete() so ignore
+    # type error
+    def delete(  # type: ignore
+        self, file_path: str, branch: str, commit_message: str, **kwargs: Any
+    ) -> None:
         """Delete a file on the server.
 
         Args:
@@ -175,8 +210,14 @@ def delete(self, file_path, branch, commit_message, **kwargs):
     @cli.register_custom_action("ProjectFileManager", ("file_path", "ref"))
     @exc.on_http_error(exc.GitlabGetError)
     def raw(
-        self, file_path, ref, streamed=False, action=None, chunk_size=1024, **kwargs
-    ):
+        self,
+        file_path: str,
+        ref: str,
+        streamed: bool = False,
+        action: Optional[Callable[..., Any]] = None,
+        chunk_size: int = 1024,
+        **kwargs: Any,
+    ) -> Optional[bytes]:
         """Return the content of a file for a commit.
 
         Args:
@@ -203,11 +244,13 @@ def raw(
         result = self.gitlab.http_get(
             path, query_data=query_data, streamed=streamed, raw=True, **kwargs
         )
+        if TYPE_CHECKING:
+            assert isinstance(result, requests.Response)
         return utils.response_content(result, streamed, action, chunk_size)
 
     @cli.register_custom_action("ProjectFileManager", ("file_path", "ref"))
     @exc.on_http_error(exc.GitlabListError)
-    def blame(self, file_path, ref, **kwargs):
+    def blame(self, file_path: str, ref: str, **kwargs: Any) -> List[Dict[str, Any]]:
         """Return the content of a file for a commit.
 
         Args:
@@ -225,4 +268,7 @@ def blame(self, file_path, ref, **kwargs):
         file_path = file_path.replace("/", "%2F").replace(".", "%2E")
         path = f"{self.path}/{file_path}/blame"
         query_data = {"ref": ref}
-        return self.gitlab.http_list(path, query_data, **kwargs)
+        result = self.gitlab.http_list(path, query_data, **kwargs)
+        if TYPE_CHECKING:
+            assert isinstance(result, list)
+        return result
diff --git a/gitlab/v4/objects/geo_nodes.py b/gitlab/v4/objects/geo_nodes.py
@@ -1,3 +1,5 @@
+from typing import Any, cast, Dict, List, TYPE_CHECKING, Union
+
 from gitlab import cli
 from gitlab import exceptions as exc
 from gitlab.base import RequiredOptional, RESTManager, RESTObject
@@ -18,7 +20,7 @@
 class GeoNode(SaveMixin, ObjectDeleteMixin, RESTObject):
     @cli.register_custom_action("GeoNode")
     @exc.on_http_error(exc.GitlabRepairError)
-    def repair(self, **kwargs):
+    def repair(self, **kwargs: Any) -> None:
         """Repair the OAuth authentication of the geo node.
 
         Args:
@@ -30,11 +32,13 @@ def repair(self, **kwargs):
         """
         path = f"/geo_nodes/{self.get_id()}/repair"
         server_data = self.manager.gitlab.http_post(path, **kwargs)
+        if TYPE_CHECKING:
+            assert isinstance(server_data, dict)
         self._update_attrs(server_data)
 
     @cli.register_custom_action("GeoNode")
     @exc.on_http_error(exc.GitlabGetError)
-    def status(self, **kwargs):
+    def status(self, **kwargs: Any) -> Dict[str, Any]:
         """Get the status of the geo node.
 
         Args:
@@ -48,7 +52,10 @@ def status(self, **kwargs):
             dict: The status of the geo node
         """
         path = f"/geo_nodes/{self.get_id()}/status"
-        return self.manager.gitlab.http_get(path, **kwargs)
+        result = self.manager.gitlab.http_get(path, **kwargs)
+        if TYPE_CHECKING:
+            assert isinstance(result, dict)
+        return result
 
 
 class GeoNodeManager(RetrieveMixin, UpdateMixin, DeleteMixin, RESTManager):
@@ -58,9 +65,12 @@ class GeoNodeManager(RetrieveMixin, UpdateMixin, DeleteMixin, RESTManager):
         optional=("enabled", "url", "files_max_capacity", "repos_max_capacity"),
     )
 
+    def get(self, id: Union[str, int], lazy: bool = False, **kwargs: Any) -> GeoNode:
+        return cast(GeoNode, super().get(id=id, lazy=lazy, **kwargs))
+
     @cli.register_custom_action("GeoNodeManager")
     @exc.on_http_error(exc.GitlabGetError)
-    def status(self, **kwargs):
+    def status(self, **kwargs: Any) -> List[Dict[str, Any]]:
         """Get the status of all the geo nodes.
 
         Args:
@@ -73,11 +83,14 @@ def status(self, **kwargs):
         Returns:
             list: The status of all the geo nodes
         """
-        return self.gitlab.http_list("/geo_nodes/status", **kwargs)
+        result = self.gitlab.http_list("/geo_nodes/status", **kwargs)
+        if TYPE_CHECKING:
+            assert isinstance(result, list)
+        return result
 
     @cli.register_custom_action("GeoNodeManager")
     @exc.on_http_error(exc.GitlabGetError)
-    def current_failures(self, **kwargs):
+    def current_failures(self, **kwargs: Any) -> List[Dict[str, Any]]:
         """Get the list of failures on the current geo node.
 
         Args:
@@ -90,4 +103,7 @@ def current_failures(self, **kwargs):
         Returns:
             list: The list of failures
         """
-        return self.gitlab.http_list("/geo_nodes/current/failures", **kwargs)
+        result = self.gitlab.http_list("/geo_nodes/current/failures", **kwargs)
+        if TYPE_CHECKING:
+            assert isinstance(result, list)
+        return result