1 files changed, 114 insertions, 0 deletions
diff --git a/gitlab/v4/objects/packages.py b/gitlab/v4/objects/packages.py
index f5ca081..a470a94 100644
--- a/gitlab/v4/objects/packages.py
+++ b/gitlab/v4/objects/packages.py
@@ -1,7 +1,17 @@
+from pathlib import Path
+from typing import Any, Callable, Optional, TYPE_CHECKING, Union
+
+import requests
+
+from gitlab import cli
+from gitlab import exceptions as exc
+from gitlab import utils
 from gitlab.base import RESTManager, RESTObject
 from gitlab.mixins import DeleteMixin, GetMixin, ListMixin, ObjectDeleteMixin
 
 __all__ = [
+    "GenericPackage",
+    "GenericPackageManager",
     "GroupPackage",
     "GroupPackageManager",
     "ProjectPackage",
@@ -11,6 +21,110 @@ __all__ = [
 ]
 
 
+class GenericPackage(RESTObject):
+    _id_attr = "package_name"
+
+
+class GenericPackageManager(RESTManager):
+    _path = "/projects/%(project_id)s/packages/generic"
+    _obj_cls = GenericPackage
+    _from_parent_attrs = {"project_id": "id"}
+
+    @cli.register_custom_action(
+        "GenericPackageManager",
+        ("package_name", "package_version", "file_name", "path"),
+    )
+    @exc.on_http_error(exc.GitlabUploadError)
+    def upload(
+        self,
+        package_name: str,
+        package_version: str,
+        file_name: str,
+        path: Union[str, Path],
+        **kwargs,
+    ) -> GenericPackage:
+        """Upload a file as a generic package.
+
+        Args:
+            package_name (str): The package name. Must follow generic package
+                                name regex rules
+            package_version (str): The package version. Must follow semantic
+                                version regex rules
+            file_name (str): The name of the file as uploaded in the registry
+            path (str): The path to a local file to upload
+
+        Raises:
+            GitlabConnectionError: If the server cannot be reached
+            GitlabUploadError: If the file upload fails
+            GitlabUploadError: If ``filepath`` cannot be read
+
+        Returns:
+            GenericPackage: An object storing the metadata of the uploaded package.
+        """
+
+        try:
+            with open(path, "rb") as f:
+                file_data = f.read()
+        except OSError:
+            raise exc.GitlabUploadError(f"Failed to read package file {path}")
+
+        url = f"{self._computed_path}/{package_name}/{package_version}/{file_name}"
+        server_data = self.gitlab.http_put(url, post_data=file_data, raw=True, **kwargs)
+
+        return self._obj_cls(
+            self,
+            {
+                "package_name": package_name,
+                "package_version": package_version,
+                "file_name": file_name,
+                "path": path,
+                "message": server_data["message"],
+            },
+        )
+
+    @cli.register_custom_action(
+        "GenericPackageManager",
+        ("package_name", "package_version", "file_name"),
+    )
+    @exc.on_http_error(exc.GitlabGetError)
+    def download(
+        self,
+        package_name: str,
+        package_version: str,
+        file_name: str,
+        streamed: bool = False,
+        action: Optional[Callable] = None,
+        chunk_size: int = 1024,
+        **kwargs: Any,
+    ) -> Optional[bytes]:
+        """Download a generic package.
+
+        Args:
+            package_name (str): The package name.
+            package_version (str): The package version.
+            file_name (str): The name of the file in the registry
+            streamed (bool): If True the data will be processed by chunks of
+                `chunk_size` and each chunk is passed to `action` for
+                reatment
+            action (callable): Callable responsible of dealing with chunk of
+                data
+            chunk_size (int): 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 server failed to perform the request
+
+        Returns:
+            str: The package content if streamed is False, None otherwise
+        """
+        path = f"{self._computed_path}/{package_name}/{package_version}/{file_name}"
+        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)
+
+
 class GroupPackage(RESTObject):
     pass