feat(docker/api/container): add support for subpath in volume_opts

TESTED: Yes, added unit tests to verify subpath functionality
Signed-off-by: Khushiyant <khushiyant2002@gmail.com>

Author: Khushiyant

.vscode/launch.json

@@ -0,0 +1,23 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python Debugger: pytest",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "${workspaceFolder}/.venv/bin/pytest",
+            "console": "integratedTerminal",
+            "args": [
+                // "--maxfail=1",
+                "--disable-warnings",
+                "-vv",
+                "${file}::VolumeBindTest::test_create_with_subpath_volume_mount"
+
+            ],
+            "justMyCode": false
+        }
+    ]
+}

docker/api/volume.py

@@ -46,6 +46,7 @@ def create_volume(self, name=None, driver=None, driver_opts=None,
             driver_opts (dict): Driver options as a key-value dictionary
             labels (dict): Labels to set on the volume
 
+
         Returns:
             (dict): The created volume reference object
 

docker/types/services.py

@@ -242,14 +242,16 @@ class Mount(dict):
           for the ``volume`` type.
         driver_config (DriverConfig): Volume driver configuration. Only valid
           for the ``volume`` type.
+        subpath (string): The path within the volume where the container's
+          volume should be mounted.
         tmpfs_size (int or string): The size for the tmpfs mount in bytes.
         tmpfs_mode (int): The permission mode for the tmpfs mount.
     """
 
     def __init__(self, target, source, type='volume', read_only=False,
                  consistency=None, propagation=None, no_copy=False,
                  labels=None, driver_config=None, tmpfs_size=None,
-                 tmpfs_mode=None):
+                 tmpfs_mode=None, subpath=None):
         self['Target'] = target
         self['Source'] = source
         if type not in ('bind', 'volume', 'tmpfs', 'npipe'):
@@ -267,7 +269,7 @@ def __init__(self, target, source, type='volume', read_only=False,
                 self['BindOptions'] = {
                     'Propagation': propagation
                 }
-            if any([labels, driver_config, no_copy, tmpfs_size, tmpfs_mode]):
+            if any([labels, driver_config, no_copy, tmpfs_size, tmpfs_mode, subpath]):
                 raise errors.InvalidArgument(
                     'Incompatible options have been provided for the bind '
                     'type mount.'
@@ -280,6 +282,8 @@ def __init__(self, target, source, type='volume', read_only=False,
                 volume_opts['Labels'] = labels
             if driver_config:
                 volume_opts['DriverConfig'] = driver_config
+            if subpath:
+                volume_opts['Subpath'] = subpath
             if volume_opts:
                 self['VolumeOptions'] = volume_opts
             if any([propagation, tmpfs_size, tmpfs_mode]):
@@ -299,7 +303,7 @@ def __init__(self, target, source, type='volume', read_only=False,
                 tmpfs_opts['SizeBytes'] = parse_bytes(tmpfs_size)
             if tmpfs_opts:
                 self['TmpfsOptions'] = tmpfs_opts
-            if any([propagation, labels, driver_config, no_copy]):
+            if any([propagation, labels, driver_config, no_copy, subpath]):
                 raise errors.InvalidArgument(
                     'Incompatible options have been provided for the tmpfs '
                     'type mount.'

tests/integration/api_container_test.py

@@ -620,6 +620,56 @@ def test_create_with_volume_mount(self):
         assert mount['Source'] == mount_data['Name']
         assert mount_data['RW'] is True
 
+    @requires_api_version('1.45')
+    def test_create_with_subpath_volume_mount(self):
+        source_volume = helpers.random_name()
+        self.client.create_volume(name=source_volume)
+
+        setup_container = None
+        test_container = None
+
+
+        # Create a file structure in the volume to test with
+        setup_container = self.client.create_container(
+            TEST_IMG,
+            [
+                "sh",
+                "-c",
+                'mkdir -p /vol/subdir && echo "test content" > /vol/subdir/testfile.txt',
+            ],
+            host_config=self.client.create_host_config(
+                binds=[f"{source_volume}:/vol"]
+            ),
+        )
+        self.client.start(setup_container)
+        self.client.wait(setup_container)
+
+        # Now test with subpath
+        mount = docker.types.Mount(
+            type="volume",
+            source=source_volume,
+            target=self.mount_dest,
+            read_only=True,
+            subpath="subdir",
+        )
+
+
+        host_config = self.client.create_host_config(mounts=[mount])
+        test_container = self.client.create_container(
+            TEST_IMG,
+            ["cat", os.path.join(self.mount_dest, "testfile.txt")],
+            host_config=host_config,
+        )
+
+        self.client.start(test_container)
+        self.client.wait(test_container)  # Wait for container to finish
+        output = self.client.logs(test_container).decode("utf-8").strip()
+
+        # If the subpath feature is working, we should be able to see the content
+        # of the file in the subdir
+        assert output == "test content"
+
+
     def check_container_data(self, inspect_data, rw, propagation='rprivate'):
         assert 'Mounts' in inspect_data
         filtered = list(filter(