Skip to content

chore: add type-hints to gitlab/v4/objects/projects.py #1511

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jun 13, 2021
Merged

chore: add type-hints to gitlab/v4/objects/projects.py #1511

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .mypy.ini
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
[mypy]
files = gitlab/*.py,gitlab/v4/cli.py
files = gitlab/*.py,gitlab/v4/cli.py,gitlab/v4/objects/projects.py

# disallow_incomplete_defs: This flag reports an error whenever it encounters a
# partly annotated function definition.
Expand Down
163 changes: 114 additions & 49 deletions gitlab/v4/objects/projects.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
from gitlab import cli
from typing import Any, Callable, cast, Dict, List, Optional, TYPE_CHECKING, Union

import requests

from gitlab import cli, client
from gitlab import exceptions as exc
from gitlab import types, utils
from gitlab.base import RequiredOptional, RESTManager, RESTObject
Expand Down Expand Up @@ -163,7 +167,7 @@ class Project(RefreshMixin, SaveMixin, ObjectDeleteMixin, RepositoryMixin, RESTO

@cli.register_custom_action("Project", ("forked_from_id",))
@exc.on_http_error(exc.GitlabCreateError)
def create_fork_relation(self, forked_from_id, **kwargs):
def create_fork_relation(self, forked_from_id: int, **kwargs: Any) -> None:
"""Create a forked from/to relation between existing projects.

Args:
Expand All @@ -179,7 +183,7 @@ def create_fork_relation(self, forked_from_id, **kwargs):

@cli.register_custom_action("Project")
@exc.on_http_error(exc.GitlabDeleteError)
def delete_fork_relation(self, **kwargs):
def delete_fork_relation(self, **kwargs: Any) -> None:
"""Delete a forked relation between existing projects.

Args:
Expand All @@ -194,7 +198,7 @@ def delete_fork_relation(self, **kwargs):

@cli.register_custom_action("Project")
@exc.on_http_error(exc.GitlabGetError)
def languages(self, **kwargs):
def languages(self, **kwargs: Any) -> Union[Dict[str, Any], requests.Response]:
"""Get languages used in the project with percentage value.

Args:
Expand All @@ -209,7 +213,7 @@ def languages(self, **kwargs):

@cli.register_custom_action("Project")
@exc.on_http_error(exc.GitlabCreateError)
def star(self, **kwargs):
def star(self, **kwargs: Any) -> None:
"""Star a project.

Args:
Expand All @@ -221,11 +225,13 @@ def star(self, **kwargs):
"""
path = "/projects/%s/star" % self.get_id()
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("Project")
@exc.on_http_error(exc.GitlabDeleteError)
def unstar(self, **kwargs):
def unstar(self, **kwargs: Any) -> None:
"""Unstar a project.

Args:
Expand All @@ -237,11 +243,13 @@ def unstar(self, **kwargs):
"""
path = "/projects/%s/unstar" % self.get_id()
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("Project")
@exc.on_http_error(exc.GitlabCreateError)
def archive(self, **kwargs):
def archive(self, **kwargs: Any) -> None:
"""Archive a project.

Args:
Expand All @@ -253,11 +261,13 @@ def archive(self, **kwargs):
"""
path = "/projects/%s/archive" % self.get_id()
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("Project")
@exc.on_http_error(exc.GitlabDeleteError)
def unarchive(self, **kwargs):
def unarchive(self, **kwargs: Any) -> None:
"""Unarchive a project.

Args:
Expand All @@ -269,13 +279,21 @@ def unarchive(self, **kwargs):
"""
path = "/projects/%s/unarchive" % self.get_id()
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(
"Project", ("group_id", "group_access"), ("expires_at",)
)
@exc.on_http_error(exc.GitlabCreateError)
def share(self, group_id, group_access, expires_at=None, **kwargs):
def share(
self,
group_id: int,
group_access: int,
expires_at: Optional[str] = None,
**kwargs: Any
) -> None:
"""Share the project with a group.

Args:
Expand All @@ -297,7 +315,7 @@ def share(self, group_id, group_access, expires_at=None, **kwargs):

@cli.register_custom_action("Project", ("group_id",))
@exc.on_http_error(exc.GitlabDeleteError)
def unshare(self, group_id, **kwargs):
def unshare(self, group_id: int, **kwargs: Any) -> None:
"""Delete a shared project link within a group.

Args:
Expand All @@ -314,7 +332,13 @@ def unshare(self, group_id, **kwargs):
# variables not supported in CLI
@cli.register_custom_action("Project", ("ref", "token"))
@exc.on_http_error(exc.GitlabCreateError)
def trigger_pipeline(self, ref, token, variables=None, **kwargs):
def trigger_pipeline(
self,
ref: str,
token: str,
variables: Optional[Dict[str, Any]] = None,
**kwargs: Any
) -> ProjectPipeline:
"""Trigger a CI build.

See https://gitlab.com/help/ci/triggers/README.md#trigger-a-build
Expand All @@ -333,11 +357,13 @@ def trigger_pipeline(self, ref, token, variables=None, **kwargs):
path = "/projects/%s/trigger/pipeline" % self.get_id()
post_data = {"ref": ref, "token": token, "variables": variables}
attrs = self.manager.gitlab.http_post(path, post_data=post_data, **kwargs)
if TYPE_CHECKING:
assert isinstance(attrs, dict)
return ProjectPipeline(self.pipelines, attrs)

@cli.register_custom_action("Project")
@exc.on_http_error(exc.GitlabHousekeepingError)
def housekeeping(self, **kwargs):
def housekeeping(self, **kwargs: Any) -> None:
"""Start the housekeeping task.

Args:
Expand All @@ -354,7 +380,13 @@ def housekeeping(self, **kwargs):
# see #56 - add file attachment features
@cli.register_custom_action("Project", ("filename", "filepath"))
@exc.on_http_error(exc.GitlabUploadError)
def upload(self, filename, filedata=None, filepath=None, **kwargs):
def upload(
self,
filename: str,
filedata: Optional[bytes] = None,
filepath: Optional[str] = None,
**kwargs: Any
) -> Dict[str, Any]:
"""Upload the specified file into the project.

.. note::
Expand Down Expand Up @@ -394,13 +426,20 @@ def upload(self, filename, filedata=None, filepath=None, **kwargs):
file_info = {"file": (filename, filedata)}
data = self.manager.gitlab.http_post(url, files=file_info)

if TYPE_CHECKING:
assert isinstance(data, dict)
return {"alt": data["alt"], "url": data["url"], "markdown": data["markdown"]}

@cli.register_custom_action("Project", optional=("wiki",))
@exc.on_http_error(exc.GitlabGetError)
def snapshot(
self, wiki=False, streamed=False, action=None, chunk_size=1024, **kwargs
):
self,
wiki: bool = False,
streamed: bool = False,
action: Optional[Callable] = None,
chunk_size: int = 1024,
**kwargs: Any
) -> Optional[bytes]:
"""Return a snapshot of the repository.

Args:
Expand All @@ -424,11 +463,15 @@ def snapshot(
result = self.manager.gitlab.http_get(
path, 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("Project", ("scope", "search"))
@exc.on_http_error(exc.GitlabSearchError)
def search(self, scope, search, **kwargs):
def search(
self, scope: str, search: str, **kwargs: Any
) -> Union[client.GitlabList, List[Dict[str, Any]]]:
"""Search the project resources matching the provided string.'

Args:
Expand All @@ -449,7 +492,7 @@ def search(self, scope, search, **kwargs):

@cli.register_custom_action("Project")
@exc.on_http_error(exc.GitlabCreateError)
def mirror_pull(self, **kwargs):
def mirror_pull(self, **kwargs: Any) -> None:
"""Start the pull mirroring process for the project.

Args:
Expand All @@ -464,7 +507,7 @@ def mirror_pull(self, **kwargs):

@cli.register_custom_action("Project", ("to_namespace",))
@exc.on_http_error(exc.GitlabTransferProjectError)
def transfer_project(self, to_namespace, **kwargs):
def transfer_project(self, to_namespace: str, **kwargs: Any) -> None:
"""Transfer a project to the given namespace ID

Args:
Expand All @@ -484,8 +527,14 @@ def transfer_project(self, to_namespace, **kwargs):
@cli.register_custom_action("Project", ("ref_name", "job"), ("job_token",))
@exc.on_http_error(exc.GitlabGetError)
def artifacts(
self, ref_name, job, streamed=False, action=None, chunk_size=1024, **kwargs
):
self,
ref_name: str,
job: str,
streamed: bool = False,
action: Optional[Callable] = None,
chunk_size: int = 1024,
**kwargs: Any
) -> Optional[bytes]:
"""Get the job artifacts archive from a specific tag or branch.

Args:
Expand Down Expand Up @@ -513,20 +562,22 @@ def artifacts(
result = self.manager.gitlab.http_get(
path, job=job, 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("Project", ("ref_name", "artifact_path", "job"))
@exc.on_http_error(exc.GitlabGetError)
def artifact(
self,
ref_name,
artifact_path,
job,
streamed=False,
action=None,
chunk_size=1024,
**kwargs
):
ref_name: str,
artifact_path: str,
job: str,
streamed: bool = False,
action: Optional[Callable] = None,
chunk_size: int = 1024,
**kwargs: Any
) -> Optional[bytes]:
"""Download a single artifact file from a specific tag or branch from within the job’s artifacts archive.

Args:
Expand Down Expand Up @@ -558,6 +609,8 @@ def artifact(
result = self.manager.gitlab.http_get(
path, streamed=streamed, raw=True, **kwargs
)
if TYPE_CHECKING:
assert isinstance(result, requests.Response)
return utils.response_content(result, streamed, action, chunk_size)


Expand Down Expand Up @@ -725,16 +778,19 @@ class ProjectManager(CRUDMixin, RESTManager):
)
_types = {"avatar": types.ImageAttribute, "topic": types.ListAttribute}

def get(self, id: Union[str, int], lazy: bool = False, **kwargs: Any) -> Project:
return cast(Project, super().get(id=id, lazy=lazy, **kwargs))

def import_project(
self,
file,
path,
name=None,
namespace=None,
overwrite=False,
override_params=None,
**kwargs
):
file: str,
path: str,
name: Optional[str] = None,
namespace: Optional[str] = None,
overwrite: bool = False,
override_params: Optional[Dict[str, Any]] = None,
**kwargs: Any
) -> Union[Dict[str, Any], requests.Response]:
"""Import a project from an archive file.

Args:
Expand Down Expand Up @@ -769,15 +825,15 @@ def import_project(

def import_bitbucket_server(
self,
bitbucket_server_url,
bitbucket_server_username,
personal_access_token,
bitbucket_server_project,
bitbucket_server_repo,
new_name=None,
target_namespace=None,
**kwargs
):
bitbucket_server_url: str,
bitbucket_server_username: str,
personal_access_token: str,
bitbucket_server_project: str,
bitbucket_server_repo: str,
new_name: Optional[str] = None,
target_namespace: Optional[str] = None,
**kwargs: Any
) -> Union[Dict[str, Any], requests.Response]:
"""Import a project from BitBucket Server to Gitlab (schedule the import)

This method will return when an import operation has been safely queued,
Expand Down Expand Up @@ -856,8 +912,13 @@ def import_bitbucket_server(
return result

def import_github(
self, personal_access_token, repo_id, target_namespace, new_name=None, **kwargs
):
self,
personal_access_token: str,
repo_id: int,
target_namespace: str,
new_name: Optional[str] = None,
**kwargs: Any
) -> Union[Dict[str, Any], requests.Response]:
"""Import a project from Github to Gitlab (schedule the import)

This method will return when an import operation has been safely queued,
Expand Down Expand Up @@ -944,7 +1005,9 @@ class ProjectForkManager(CreateMixin, ListMixin, RESTManager):
)
_create_attrs = RequiredOptional(optional=("namespace",))

def create(self, data, **kwargs):
def create(
self, data: Optional[Dict[str, Any]] = None, **kwargs: Any
) -> ProjectFork:
"""Creates a new object.

Args:
Expand All @@ -960,8 +1023,10 @@ def create(self, data, **kwargs):
RESTObject: A new instance of the managed object class build with
the data sent by the server
"""
if TYPE_CHECKING:
assert self.path is not None
path = self.path[:-1] # drop the 's'
return CreateMixin.create(self, data, path=path, **kwargs)
return cast(ProjectFork, CreateMixin.create(self, data, path=path, **kwargs))


class ProjectRemoteMirror(SaveMixin, RESTObject):
Expand Down