Jun 26, 2022 · Mar 30, 2022 · Apr 9, 2022 · Apr 9, 2022 · Apr 10, 2022 · Apr 10, 2022
diff --git a/docs/gl_objects/pipelines_and_jobs.rst b/docs/gl_objects/pipelines_and_jobs.rst
    subprocess.run(["unzip", "-bo", zipfn])
    os.unlink(zipfn)

 Or, you can also use the underlying response iterator directly::

    artifact_bytes_iterator = build_or_job.artifacts(iterator=True)

 This can be used with frameworks that expect an iterator (such as FastAPI/Starlette's
 ``StreamingResponse``) to forward a download from GitLab without having to download
 the entire content server-side first::

    @app.get("/download_artifact")
    def download_artifact():
        artifact_bytes_iterator = build_or_job.artifacts(iterator=True)
        return StreamingResponse(artifact_bytes_iterator, media_type="application/zip")

 Delete all artifacts of a project that can be deleted::

  project.artifacts.delete()
diff --git a/gitlab/mixins.py b/gitlab/mixins.py
    Any,
    Callable,
    Dict,
    Iterator,
    List,
    Optional,
    Tuple,
    def download(
        self,
        streamed: bool = False,
        iterator: bool = False,
        action: Optional[Callable] = None,
        chunk_size: int = 1024,
        **kwargs: Any,
    ) -> Optional[bytes]:
    ) -> Optional[Union[bytes, Iterator[Any]]]:
        """Download the archive of a resource export.

        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
        )
        if TYPE_CHECKING:
            assert isinstance(result, requests.Response)
        return utils.response_content(result, streamed, action, chunk_size)
        return utils.response_content(result, streamed,iterator,action, chunk_size)


 class SubscribableMixin(_RestObjectBase):
diff --git a/gitlab/utils.py b/gitlab/utils.py
 import traceback
 import urllib.parse
 import warnings
 from typing import Any, Callable, Dict, Optional, Tuple, Type, Union
 from typing import Any, Callable, Dict,Iterator,Optional, Tuple, Type, Union

 import requests

 def response_content(
    response: requests.Response,
    streamed: bool,
    iterator: bool,
    action: Optional[Callable],
    chunk_size: int,
 ) -> Optional[bytes]:
 ) -> Optional[Union[bytes, Iterator[Any]]]:
    if iterator:
        return response.iter_content(chunk_size=chunk_size)

    if streamed is False:
        return response.content

diff --git a/gitlab/v4/cli.py b/gitlab/v4/cli.py
            data = export_status.download()
            if TYPE_CHECKING:
                assert data is not None
                assert isinstance(data, bytes)
            sys.stdout.buffer.write(data)

        except Exception as e:  # pragma: no cover, cli.die is unit-tested
diff --git a/gitlab/v4/objects/artifacts.py b/gitlab/v4/objects/artifacts.py
 GitLab API:
 https://docs.gitlab.com/ee/api/job_artifacts.html
 """
 from typing import Any, Callable, Optional, TYPE_CHECKING
 from typing import Any, Callable,Iterator,Optional, TYPE_CHECKING, Union

 import requests

            ),
            category=DeprecationWarning,
        )
 return self.download(
 data = self.download(
            *args,
            **kwargs,
        )
        if TYPE_CHECKING:
            assert data is not None
            assert isinstance(data, bytes)
        return data

    @exc.on_http_error(exc.GitlabDeleteError)
    def delete(self, **kwargs: Any) -> None:
        ref_name: str,
        job: str,
        streamed: bool = False,
        iterator: bool = False,
        action: Optional[Callable] = None,
        chunk_size: int = 1024,
        **kwargs: Any,
    ) -> Optional[bytes]:
    ) -> Optional[Union[bytes, Iterator[Any]]]:
        """Get the job artifacts archive from a specific tag or branch.

        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
        )
        if TYPE_CHECKING:
            assert isinstance(result, requests.Response)
        return utils.response_content(result, streamed, action, chunk_size)
        return utils.response_content(result, streamed,iterator,action, chunk_size)

    @cli.register_custom_action(
        "ProjectArtifactManager", ("ref_name", "artifact_path", "job")
        artifact_path: str,
        job: str,
        streamed: bool = False,
        iterator: bool = False,
        action: Optional[Callable] = None,
        chunk_size: int = 1024,
        **kwargs: Any,
    ) -> Optional[bytes]:
    ) -> Optional[Union[bytes, Iterator[Any]]]:
        """Download a single artifact file from a specific tag or branch from
        within the job's artifacts archive.

            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
        )
        if TYPE_CHECKING:
            assert isinstance(result, requests.Response)
        return utils.response_content(result, streamed, action, chunk_size)
        return utils.response_content(result, streamed,iterator,action, chunk_size)
diff --git a/gitlab/v4/objects/files.py b/gitlab/v4/objects/files.py
 import base64
 from typing import Any, Callable, cast, Dict, List, Optional, TYPE_CHECKING
 from typing import (
    Any,
    Callable,
    cast,
    Dict,
    Iterator,
    List,
    Optional,
    TYPE_CHECKING,
    Union,
 )

 import requests

        file_path: str,
        ref: str,
        streamed: bool = False,
        iterator: bool = False,
        action: Optional[Callable[..., Any]] = None,
        chunk_size: int = 1024,
        **kwargs: Any,
    ) -> Optional[bytes]:
    ) -> Optional[Union[bytes, Iterator[Any]]]:
        """Return the content of a file for a commit.

        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
        )
        if TYPE_CHECKING:
            assert isinstance(result, requests.Response)
        return utils.response_content(result, streamed, action, chunk_size)
        return utils.response_content(result, streamed,iterator,action, chunk_size)

    @cli.register_custom_action("ProjectFileManager", ("file_path", "ref"))
    @exc.on_http_error(exc.GitlabListError)
diff --git a/gitlab/v4/objects/jobs.py b/gitlab/v4/objects/jobs.py
 from typing import Any, Callable, cast, Dict, Optional, TYPE_CHECKING, Union
 from typing import Any, Callable, cast, Dict,Iterator,Optional, TYPE_CHECKING, Union

 import requests

    def artifacts(
        self,
        streamed: bool = False,
        iterator: bool = False,
        action: Optional[Callable[..., Any]] = None,
        chunk_size: int = 1024,
        **kwargs: Any,
    ) -> Optional[bytes]:
    ) -> 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
        )
        if TYPE_CHECKING:
            assert isinstance(result, requests.Response)
        return utils.response_content(result, streamed, action, chunk_size)
        return utils.response_content(result, streamed,iterator,action, chunk_size)

    @cli.register_custom_action("ProjectJob")
    @exc.on_http_error(exc.GitlabGetError)
    def artifact(
        self,
        path: str,
        streamed: bool = False,
        iterator: bool = False,
        action: Optional[Callable[..., Any]] = None,
        chunk_size: int = 1024,
        **kwargs: Any,
    ) -> Optional[bytes]:
    ) -> 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
        )
        if TYPE_CHECKING:
            assert isinstance(result, requests.Response)
        return utils.response_content(result, streamed, action, chunk_size)
        return utils.response_content(result, streamed,iterator,action, chunk_size)

    @cli.register_custom_action("ProjectJob")
    @exc.on_http_error(exc.GitlabGetError)
    def trace(
        self,
        streamed: bool = False,
        iterator: bool = False,
        action: Optional[Callable[..., Any]] = None,
        chunk_size: int = 1024,
        **kwargs: Any,
            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
        )
        if TYPE_CHECKING:
            assert isinstance(result, requests.Response)
        return_value = utils.response_content(result, streamed, action, chunk_size)
        return_value = utils.response_content(
            result, streamed, iterator, action, chunk_size
        )
        if TYPE_CHECKING:
            assert isinstance(return_value, dict)
        return return_value
diff --git a/gitlab/v4/objects/packages.py b/gitlab/v4/objects/packages.py
 """

 from pathlib import Path
 from typing import Any, Callable, cast, Optional, TYPE_CHECKING, Union
 from typing import Any, Callable, cast,Iterator,Optional, TYPE_CHECKING, Union

 import requests

        package_version: str,
        file_name: str,
        streamed: bool = False,
        iterator: bool = False,
        action: Optional[Callable] = None,
        chunk_size: int = 1024,
        **kwargs: Any,
    ) -> Optional[bytes]:
    ) -> Optional[Union[bytes, Iterator[Any]]]:
        """Download a generic package.

        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
        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)
        return utils.response_content(result, streamed,iterator,action, chunk_size)


 class GroupPackage(RESTObject):
Original file line number	Diff line number	Diff line change
Expand Up		@@ -274,6 +274,19 @@ You can also directly stream the output into a file, and unzip it afterwards::
		subprocess.run(["unzip", "-bo", zipfn])
		os.unlink(zipfn)

		Or, you can also use the underlying response iterator directly::

		artifact_bytes_iterator = build_or_job.artifacts(iterator=True)

		This can be used with frameworks that expect an iterator (such as FastAPI/Starlette's
		``StreamingResponse``) to forward a download from GitLab without having to download
		the entire content server-side first::

		@app.get("/download_artifact")
		def download_artifact():
		artifact_bytes_iterator = build_or_job.artifacts(iterator=True)
		return StreamingResponse(artifact_bytes_iterator, media_type="application/zip")

		Delete all artifacts of a project that can be deleted::

		project.artifacts.delete()
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -20,6 +20,7 @@
		Any,
		Callable,
		Dict,
		Iterator,
		List,
		Optional,
		Tuple,
Expand DownExpand Up		@@ -614,16 +615,19 @@ class DownloadMixin(_RestObjectBase):
		def download(
		self,
		streamed: bool = False,
		iterator: bool = False,
		action: Optional[Callable] = None,
		chunk_size: int = 1024,
		**kwargs: Any,
		) -> Optional[bytes]:
		) -> Optional[Union[bytes, Iterator[Any]]]:
		"""Download the archive of a resource export.

		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
Expand All		@@ -642,7 +646,7 @@ def download(
		)
		if TYPE_CHECKING:
		assert isinstance(result, requests.Response)
		return utils.response_content(result, streamed, action, chunk_size)
		return utils.response_content(result, streamed,iterator,action, chunk_size)


		class SubscribableMixin(_RestObjectBase):
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -19,7 +19,7 @@
		import traceback
		import urllib.parse
		import warnings
		from typing import Any, Callable, Dict, Optional, Tuple, Type, Union
		from typing import Any, Callable, Dict,Iterator,Optional, Tuple, Type, Union

		import requests

Expand All		@@ -34,9 +34,13 @@ def __call__(self, chunk: Any) -> None:
		def response_content(
		response: requests.Response,
		streamed: bool,
		iterator: bool,
		action: Optional[Callable],
		chunk_size: int,
		) -> Optional[bytes]:
		) -> Optional[Union[bytes, Iterator[Any]]]:
		if iterator:
		return response.iter_content(chunk_size=chunk_size)

		if streamed is False:
		return response.content

Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -127,6 +127,7 @@ def do_project_export_download(self) -> None:
		data = export_status.download()
		if TYPE_CHECKING:
		assert data is not None
		assert isinstance(data, bytes)
		sys.stdout.buffer.write(data)

		except Exception as e: # pragma: no cover, cli.die is unit-tested
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -2,7 +2,7 @@
		GitLab API:
		https://docs.gitlab.com/ee/api/job_artifacts.html
		"""
		from typing import Any, Callable, Optional, TYPE_CHECKING
		from typing import Any, Callable,Iterator,Optional, TYPE_CHECKING, Union

		import requests

Expand DownExpand Up		@@ -40,10 +40,14 @@ def __call__(
		),
		category=DeprecationWarning,
		)
		return self.download(
		data = self.download(
		*args,
		**kwargs,
		)
		if TYPE_CHECKING:
		assert data is not None
		assert isinstance(data, bytes)
		return data

		@exc.on_http_error(exc.GitlabDeleteError)
		def delete(self, **kwargs: Any) -> None:
Expand DownExpand Up		@@ -71,10 +75,11 @@ def download(
		ref_name: str,
		job: str,
		streamed: bool = False,
		iterator: bool = False,
		action: Optional[Callable] = None,
		chunk_size: int = 1024,
		**kwargs: Any,
		) -> Optional[bytes]:
		) -> Optional[Union[bytes, Iterator[Any]]]:
		"""Get the job artifacts archive from a specific tag or branch.

		Args:
Expand All		@@ -85,6 +90,8 @@ def download(
		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
Expand All		@@ -103,7 +110,7 @@ def download(
		)
		if TYPE_CHECKING:
		assert isinstance(result, requests.Response)
		return utils.response_content(result, streamed, action, chunk_size)
		return utils.response_content(result, streamed,iterator,action, chunk_size)

		@cli.register_custom_action(
		"ProjectArtifactManager", ("ref_name", "artifact_path", "job")
Expand All		@@ -115,10 +122,11 @@ def raw(
		artifact_path: str,
		job: str,
		streamed: bool = False,
		iterator: bool = False,
		action: Optional[Callable] = None,
		chunk_size: int = 1024,
		**kwargs: Any,
		) -> Optional[bytes]:
		) -> Optional[Union[bytes, Iterator[Any]]]:
		"""Download a single artifact file from a specific tag or branch from
		within the job's artifacts archive.

Expand All		@@ -130,6 +138,8 @@ def raw(
		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
Expand All		@@ -148,4 +158,4 @@ def raw(
		)
		if TYPE_CHECKING:
		assert isinstance(result, requests.Response)
		return utils.response_content(result, streamed, action, chunk_size)
		return utils.response_content(result, streamed,iterator,action, chunk_size)
Original file line number	Diff line number	Diff line change
		@@ -1,5 +1,15 @@
		import base64
		from typing import Any, Callable, cast, Dict, List, Optional, TYPE_CHECKING
		from typing import (
		Any,
		Callable,
		cast,
		Dict,
		Iterator,
		List,
		Optional,
		TYPE_CHECKING,
		Union,
		)

		import requests

Expand DownExpand Up		@@ -220,10 +230,11 @@ def raw(
		file_path: str,
		ref: str,
		streamed: bool = False,
		iterator: bool = False,
		action: Optional[Callable[..., Any]] = None,
		chunk_size: int = 1024,
		**kwargs: Any,
		) -> Optional[bytes]:
		) -> Optional[Union[bytes, Iterator[Any]]]:
		"""Return the content of a file for a commit.

		Args:
Expand All		@@ -232,6 +243,8 @@ def raw(
		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
Expand All		@@ -252,7 +265,7 @@ def raw(
		)
		if TYPE_CHECKING:
		assert isinstance(result, requests.Response)
		return utils.response_content(result, streamed, action, chunk_size)
		return utils.response_content(result, streamed,iterator,action, chunk_size)

		@cli.register_custom_action("ProjectFileManager", ("file_path", "ref"))
		@exc.on_http_error(exc.GitlabListError)
Expand Down
Original file line number	Diff line number	Diff line change
		@@ -1,4 +1,4 @@
		from typing import Any, Callable, cast, Dict, Optional, TYPE_CHECKING, Union
		from typing import Any, Callable, cast, Dict,Iterator,Optional, TYPE_CHECKING, Union

		import requests

Expand DownExpand Up		@@ -116,16 +116,19 @@ def delete_artifacts(self, **kwargs: Any) -> None:
		def artifacts(
		self,
		streamed: bool = False,
		iterator: bool = False,
		action: Optional[Callable[..., Any]] = None,
		chunk_size: int = 1024,
		**kwargs: Any,
		) -> Optional[bytes]:
		) -> 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
Expand All		@@ -144,25 +147,28 @@ def artifacts(
		)
		if TYPE_CHECKING:
		assert isinstance(result, requests.Response)
		return utils.response_content(result, streamed, action, chunk_size)
		return utils.response_content(result, streamed,iterator,action, chunk_size)

		@cli.register_custom_action("ProjectJob")
		@exc.on_http_error(exc.GitlabGetError)
		def artifact(
		self,
		path: str,
		streamed: bool = False,
		iterator: bool = False,
		action: Optional[Callable[..., Any]] = None,
		chunk_size: int = 1024,
		**kwargs: Any,
		) -> Optional[bytes]:
		) -> 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
Expand All		@@ -181,13 +187,14 @@ def artifact(
		)
		if TYPE_CHECKING:
		assert isinstance(result, requests.Response)
		return utils.response_content(result, streamed, action, chunk_size)
		return utils.response_content(result, streamed,iterator,action, chunk_size)

		@cli.register_custom_action("ProjectJob")
		@exc.on_http_error(exc.GitlabGetError)
		def trace(
		self,
		streamed: bool = False,
		iterator: bool = False,
		action: Optional[Callable[..., Any]] = None,
		chunk_size: int = 1024,
		**kwargs: Any,
Expand All		@@ -198,6 +205,8 @@ def trace(
		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
Expand All		@@ -216,7 +225,9 @@ def trace(
		)
		if TYPE_CHECKING:
		assert isinstance(result, requests.Response)
		return_value = utils.response_content(result, streamed, action, chunk_size)
		return_value = utils.response_content(
		result, streamed, iterator, action, chunk_size
		)
		if TYPE_CHECKING:
		assert isinstance(return_value, dict)
		return return_value
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -5,7 +5,7 @@
		"""

		from pathlib import Path
		from typing import Any, Callable, cast, Optional, TYPE_CHECKING, Union
		from typing import Any, Callable, cast,Iterator,Optional, TYPE_CHECKING, Union

		import requests

Expand DownExpand Up		@@ -103,10 +103,11 @@ def download(
		package_version: str,
		file_name: str,
		streamed: bool = False,
		iterator: bool = False,
		action: Optional[Callable] = None,
		chunk_size: int = 1024,
		**kwargs: Any,
		) -> Optional[bytes]:
		) -> Optional[Union[bytes, Iterator[Any]]]:
		"""Download a generic package.

		Args:
Expand All		@@ -116,6 +117,8 @@ def download(
		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
Expand All		@@ -132,7 +135,7 @@ def download(
		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)
		return utils.response_content(result, streamed,iterator,action, chunk_size)


		class GroupPackage(RESTObject):
Expand Down