Implement client.quadlets.install

podman/api/client.py

@@ -317,6 +317,7 @@ def post(
         *,
         params: _Params = None,
         data: _Data = None,
+        files: Optional[Any] = None,
         headers: Optional[Mapping[str, str]] = None,
         timeout: _Timeout = None,
         stream: Optional[bool] = False,
@@ -327,6 +328,8 @@ def post(
         Args:
             path: Relative path to RESTful resource.
             data: HTTP body for operation
+            files: Dictionary or list of tuples for multipart file upload.
+                Follows the ``requests`` library ``files`` parameter format.
             params: Optional parameters to include with URL.
             headers: Optional headers to include in request.
             timeout: Number of seconds to wait on request, or (connect timeout, read timeout) tuple
@@ -344,6 +347,7 @@ def post(
             path=path,
             params=params,
             data=data,
+            files=files,
             headers=headers,
             timeout=timeout,
             stream=stream,
@@ -394,6 +398,7 @@ def _request(
         path: Union[str, bytes],
         *,
         data: _Data = None,
+        files: Optional[Any] = None,
         params: _Params = None,
         headers: Optional[Mapping[str, str]] = None,
         timeout: _Timeout = None,
@@ -406,6 +411,7 @@ def _request(
             method: HTTP method to use for request
             path: Relative path to RESTful resource.
             params: Optional parameters to include with URL.
+            files: Dictionary or list of tuples for multipart file upload.
             headers: Optional headers to include in request.
             timeout: Number of seconds to wait on request, or (connect timeout, read timeout) tuple
 
@@ -445,6 +451,7 @@ def _request(
                     uri.geturl(),
                     params=params,
                     data=data,
+                    files=files,
                     headers=(headers or {}),
                     stream=stream,
                     verify=kwargs.get("verify", None),

podman/domain/quadlets.py

@@ -2,7 +2,9 @@
 
 import builtins
 import logging
-from typing import Optional, Union
+import os
+import pathlib
+from typing import Any, Optional, Union
 
 import requests
 
@@ -223,3 +225,118 @@ def delete(
 
         response.raise_for_status()
         return response.json()["Removed"]
+
+    def install(
+        self,
+        files: Union[
+            "QuadletFileItem",
+            builtins.list["QuadletFileItem"],
+        ],
+        *,
+        replace: bool = False,
+        reload_systemd: bool = True,
+    ) -> dict[str, Any]:
+        """Install a Quadlet file and additional asset files.
+
+        The function will make a single request. Each request must contain exactly
+        one quadlet file (identified by its extension: .container, .volume,
+        .network, ... etc.) and may optionally include asset files such as
+        Containerfiles, kube YAML, or other configuration files.
+
+        Quadlets and asset files can be provided as a tuple (filename, content)
+        or a string/path that represents a file path. Both can be combined in a
+        list to be included as part of the same request.
+
+        The path to a single ``.tar`` file can also be provided; it will be posted
+        directly with ``Content-Type: application/x-tar``.  In all other cases the
+        items are uploaded as ``multipart/form-data``.
+
+        Args:
+            files (Union[QuadletFileItem, list[QuadletFileItem]]): File(s) to install.
+                QuadletFileItem is a Union[tuple[str, str], str, os.PathLike].
+            replace (bool): Replace existing files if they already exist.
+                Defaults to False.
+            reload_systemd (bool): Reload systemd after installing quadlets.
+                Defaults to True.
+
+        Returns:
+            A dict with keys:
+                - ``InstalledQuadlets``: mapping of source path to installed
+                  path for successfully installed files.
+                - ``QuadletErrors``: mapping of source path to error message
+                  for failed installations (empty on success).
+
+        Raises:
+            APIError: when the service reports an error (e.g. no quadlet files
+                found, multiple quadlet files, file already exists without
+                replace, or other server errors).
+            FileNotFoundError: when a provided file path does not exist on
+                disk.
+        """
+        if not isinstance(files, builtins.list):
+            files = [files]
+
+        params = {
+            "replace": replace,
+            "reload-systemd": reload_systemd,
+        }
+
+        first = files[0]
+        if len(files) == 1 and isinstance(first, (str, os.PathLike)) and self._is_tar_path(first):
+            tar_path = pathlib.Path(first)
+            if not tar_path.is_file():
+                raise FileNotFoundError(f"No such file: '{tar_path}'")
+            response = self.client.post(
+                "/quadlets",
+                params=params,
+                data=tar_path.read_bytes(),
+                headers={"Content-Type": "application/x-tar"},
+            )
+        else:
+            multipart = self._prepare_install_body(files)
+            response = self.client.post(
+                "/quadlets",
+                params=params,
+                files=multipart,
+            )
+
+        response.raise_for_status()
+        return response.json()
+
+    @staticmethod
+    def _is_tar_path(item: "QuadletFileItem") -> bool:
+        """Return True if *item* looks like a path to a tar archive."""
+        if not isinstance(item, (str, os.PathLike)):
+            return False
+        name = str(item)
+        return name.endswith(".tar") or name.endswith(".tar.gz")
+
+    def _prepare_install_body(
+        self,
+        items: builtins.list["QuadletFileItem"],
+    ) -> dict[str, tuple[str, bytes]]:
+        """Build a ``files`` dict for :pymethod:`requests.Session.request`.
+
+        Returns a dictionary of ``{field_name: (filename, file_bytes)}``
+        suitable for passing as the ``files`` keyword argument to
+        :pymethod:`requests.Session.request`, which encodes them as
+        ``multipart/form-data``.
+        """
+        result: dict[str, tuple[str, bytes]] = {}
+        for item in items:
+            if isinstance(item, tuple):
+                filename, content = item
+                if isinstance(content, bytes):
+                    result[filename] = (filename, content)
+                else:
+                    result[filename] = (filename, content.encode("utf-8"))
+            elif isinstance(item, (str, os.PathLike)):
+                fp = pathlib.Path(item)
+                if not fp.is_file():
+                    raise FileNotFoundError(f"No such file: '{fp}'")
+                result[fp.name] = (fp.name, fp.read_bytes())
+        return result
+
+
+# Type alias – importable for type annotations in calling code.
+QuadletFileItem = Union[tuple[str, Union[str, bytes]], str, os.PathLike]