@@ -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) returncast(ProjectFork, CreateMixin.create(self, data, path=path, **kwargs) ) class ProjectRemoteMirror(SaveMixin, RESTObject): Expand Down