diff options
Diffstat (limited to 'gitlab/v4/objects')
| -rw-r--r-- | gitlab/v4/objects/artifacts.py | 116 | ||||
| -rw-r--r-- | gitlab/v4/objects/jobs.py | 83 |
2 files changed, 121 insertions, 78 deletions
diff --git a/gitlab/v4/objects/artifacts.py b/gitlab/v4/objects/artifacts.py index b4a4c0e..03972d9 100644 --- a/gitlab/v4/objects/artifacts.py +++ b/gitlab/v4/objects/artifacts.py @@ -2,6 +2,7 @@ GitLab API: https://docs.gitlab.com/ee/api/job_artifacts.html """ +import warnings from typing import Any, Callable, Iterator, Optional, TYPE_CHECKING, Union import requests @@ -11,7 +12,12 @@ from gitlab import exceptions as exc from gitlab import utils from gitlab.base import RESTManager, RESTObject -__all__ = ["ProjectArtifact", "ProjectArtifactManager"] +__all__ = [ + "ProjectArtifact", + "ProjectArtifactManager", + "ProjectJobArtifact", + "ProjectJobArtifactManager", +] class ProjectArtifact(RESTObject): @@ -165,3 +171,111 @@ class ProjectArtifactManager(RESTManager): return utils.response_content( result, streamed, action, chunk_size, iterator=iterator ) + + +class ProjectJobArtifact(RESTObject): + """Dummy object to manage custom actions on artifacts""" + + _id_attr = "artifact_path" + + +class ProjectJobArtifactManager(RESTManager): + _path = "/projects/{project_id}/jobs/{job_id}/artifacts" + _from_parent_attrs = {"project_id": "project_id", "job_id": "id"} + + @cli.register_custom_action("ProjectJob", custom_action="artifacts") + def __call__( + self, + *args: Any, + **kwargs: Any, + ) -> Optional[Union[bytes, Iterator[Any]]]: + warnings.warn( + "The job.artifacts() method is deprecated and will be " + "removed in a future version. Use job.artifacts.download() instead.\n", + DeprecationWarning, + ) + return self.download( + *args, + **kwargs, + ) + + @cli.register_custom_action("ProjectJobArtifactManager") + @exc.on_http_error(exc.GitlabGetError) + def download( + self, + streamed: bool = False, + action: Optional[Callable[..., Any]] = None, + chunk_size: int = 1024, + iterator: bool = False, + **kwargs: Any, + ) -> Optional[Union[bytes, Iterator[Any]]]: + """Get the job artifacts. + + Args: + streamed: If True the data will be processed by chunks of + `chunk_size` and each chunk is passed to `action` for + treatment + action: Callable responsible of dealing with chunk of + data + chunk_size: Size of each chunk + iterator: If True directly return the underlying response + iterator + **kwargs: Extra options to send to the server (e.g. sudo) + + Raises: + GitlabAuthenticationError: If authentication is not correct + GitlabGetError: If the artifacts could not be retrieved + + Returns: + The artifacts if `streamed` is False, None otherwise. + """ + if TYPE_CHECKING: + assert self.path is not None + result = self.gitlab.http_get(self.path, streamed=streamed, raw=True, **kwargs) + + if TYPE_CHECKING: + assert isinstance(result, requests.Response) + return utils.response_content( + result, streamed, action, chunk_size, iterator=iterator + ) + + @cli.register_custom_action("ProjectJobArtifactManager", ("artifact_path",)) + @cli.register_custom_action("ProjectJob") + @exc.on_http_error(exc.GitlabGetError) + def raw( + self, + path: str, + streamed: bool = False, + action: Optional[Callable[..., Any]] = None, + chunk_size: int = 1024, + iterator: bool = False, + **kwargs: Any, + ) -> Optional[Union[bytes, Iterator[Any]]]: + """Get a single artifact file from within the job's artifacts archive. + + Args: + path: Path of the artifact + streamed: If True the data will be processed by chunks of + `chunk_size` and each chunk is passed to `action` for + treatment + action: Callable responsible of dealing with chunk of + data + chunk_size: Size of each chunk + iterator: If True directly return the underlying response + iterator + **kwargs: Extra options to send to the server (e.g. sudo) + + Raises: + GitlabAuthenticationError: If authentication is not correct + GitlabGetError: If the artifacts could not be retrieved + + Returns: + The artifacts if `streamed` is False, None otherwise. + """ + path = f"{self.path}/{path}" + result = self.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, iterator=iterator + ) diff --git a/gitlab/v4/objects/jobs.py b/gitlab/v4/objects/jobs.py index cfe1e62..52fff50 100644 --- a/gitlab/v4/objects/jobs.py +++ b/gitlab/v4/objects/jobs.py @@ -8,6 +8,8 @@ from gitlab import utils from gitlab.base import RESTManager, RESTObject from gitlab.mixins import RefreshMixin, RetrieveMixin +from .artifacts import ProjectJobArtifactManager # noqa: F401 + __all__ = [ "ProjectJob", "ProjectJobManager", @@ -15,6 +17,8 @@ __all__ = [ class ProjectJob(RefreshMixin, RESTObject): + artifacts: ProjectJobArtifactManager + @cli.register_custom_action("ProjectJob") @exc.on_http_error(exc.GitlabJobCancelError) def cancel(self, **kwargs: Any) -> Dict[str, Any]: @@ -113,87 +117,12 @@ class ProjectJob(RefreshMixin, RESTObject): @cli.register_custom_action("ProjectJob") @exc.on_http_error(exc.GitlabGetError) - def artifacts( - self, - streamed: bool = False, - action: Optional[Callable[..., Any]] = None, - chunk_size: int = 1024, - *, - iterator: bool = False, - **kwargs: Any, - ) -> Optional[Union[bytes, Iterator[Any]]]: - """Get the job artifacts. - - Args: - streamed: If True the data will be processed by chunks of - `chunk_size` and each chunk is passed to `action` for - treatment - iterator: If True directly return the underlying response - iterator - action: Callable responsible of dealing with chunk of - data - chunk_size: Size of each chunk - **kwargs: Extra options to send to the server (e.g. sudo) - - Raises: - GitlabAuthenticationError: If authentication is not correct - GitlabGetError: If the artifacts could not be retrieved - - Returns: - The artifacts if `streamed` is False, None otherwise. - """ - path = f"{self.manager.path}/{self.encoded_id}/artifacts" - 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, iterator=iterator - ) - - @cli.register_custom_action("ProjectJob") - @exc.on_http_error(exc.GitlabGetError) def artifact( self, - path: str, - streamed: bool = False, - action: Optional[Callable[..., Any]] = None, - chunk_size: int = 1024, - *, - iterator: bool = False, + *args: Any, **kwargs: Any, ) -> Optional[Union[bytes, Iterator[Any]]]: - """Get a single artifact file from within the job's artifacts archive. - - Args: - path: Path of the artifact - streamed: If True the data will be processed by chunks of - `chunk_size` and each chunk is passed to `action` for - treatment - iterator: If True directly return the underlying response - iterator - action: Callable responsible of dealing with chunk of - data - chunk_size: Size of each chunk - **kwargs: Extra options to send to the server (e.g. sudo) - - Raises: - GitlabAuthenticationError: If authentication is not correct - GitlabGetError: If the artifacts could not be retrieved - - Returns: - The artifacts if `streamed` is False, None otherwise. - """ - path = f"{self.manager.path}/{self.encoded_id}/artifacts/{path}" - 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, iterator=iterator - ) + return self.artifacts.raw(*args, **kwargs) @cli.register_custom_action("ProjectJob") @exc.on_http_error(exc.GitlabGetError) |