Add support for MDTM command in FTPFS (#462)

Author: althonos

CHANGELOG.md

@@ -12,6 +12,13 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 - Added `fs.copy.copy_file_if`, `fs.copy.copy_dir_if`, and `fs.copy.copy_fs_if`. 
   Closes [#458](https://github.com/PyFilesystem/pyfilesystem2/issues/458).
+- Added `fs.base.FS.getmodified`.
+  
+### Changed
+
+- FTP servers that do not support the MLST command now try to use the MDTM command to
+  retrieve the last modification timestamp of a resource. 
+  Closes [#456](https://github.com/PyFilesystem/pyfilesystem2/pull/456).
 
 ### Fixed
 

docs/source/interface.rst

@@ -20,6 +20,7 @@ The following is a complete list of methods on PyFilesystem objects.
 * :meth:`~fs.base.FS.getdetails` Get details info namespace for a resource.
 * :meth:`~fs.base.FS.getinfo` Get info regarding a file or directory.
 * :meth:`~fs.base.FS.getmeta` Get meta information for a resource.
+* :meth:`~fs.base.FS.getmodified` Get info regarding the last modified time of a resource.
 * :meth:`~fs.base.FS.getospath` Get path with encoding expected by the OS.
 * :meth:`~fs.base.FS.getsize` Get the size of a file.
 * :meth:`~fs.base.FS.getsyspath` Get the system path of a resource, if one exists.

fs/base.py

@@ -697,6 +697,23 @@ def readtext(
 
     gettext = _new_name(readtext, "gettext")
 
+    def getmodified(self, path):
+        # type: (Text) -> Optional[datetime]
+        """Get the timestamp of the last modifying access of a resource.
+
+        Arguments:
+            path (str): A path to a resource.
+
+        Returns:
+            datetime: The timestamp of the last modification.
+
+        The *modified timestamp* of a file is the point in time
+        that the file was last changed. Depending on the file system,
+        it might only have limited accuracy.
+
+        """
+        return self.getinfo(path, namespaces=["details"]).modified
+
     def getmeta(self, namespace="standard"):
         # type: (Text) -> Mapping[Text, object]
         """Get meta information regarding a filesystem.

fs/copy.py

@@ -463,9 +463,8 @@ def _copy_is_necessary(
 
     elif condition == "newer":
         try:
-            namespace = ("details",)
-            src_modified = src_fs.getinfo(src_path, namespace).modified
-            dst_modified = dst_fs.getinfo(dst_path, namespace).modified
+            src_modified = src_fs.getmodified(src_path)
+            dst_modified = dst_fs.getmodified(dst_path)
         except ResourceNotFound:
             return True
         else:
@@ -477,9 +476,8 @@ def _copy_is_necessary(
 
     elif condition == "older":
         try:
-            namespace = ("details",)
-            src_modified = src_fs.getinfo(src_path, namespace).modified
-            dst_modified = dst_fs.getinfo(dst_path, namespace).modified
+            src_modified = src_fs.getmodified(src_path)
+            dst_modified = dst_fs.getmodified(dst_path)
         except ResourceNotFound:
             return True
         else:

fs/ftpfs.py

@@ -40,6 +40,7 @@
 from .path import basename
 from .path import normpath
 from .path import split
+from .time import epoch_to_datetime
 from . import _ftp_parse as ftp_parse
 
 if typing.TYPE_CHECKING:
@@ -572,6 +573,12 @@ def supports_mlst(self):
         """bool: whether the server supports MLST feature."""
         return "MLST" in self.features
 
+    @property
+    def supports_mdtm(self):
+        # type: () -> bool
+        """bool: whether the server supports the MDTM feature."""
+        return "MDTM" in self.features
+
     def create(self, path, wipe=False):
         # type: (Text, bool) -> bool
         _path = self.validatepath(path)
@@ -692,8 +699,21 @@ def getmeta(self, namespace="standard"):
         if namespace == "standard":
             _meta = self._meta.copy()
             _meta["unicode_paths"] = "UTF8" in self.features
+            _meta["supports_mtime"] = "MDTM" in self.features
         return _meta
 
+    def getmodified(self, path):
+        # type: (Text) -> Optional[datetime.datetime]
+        if self.supports_mdtm:
+            _path = self.validatepath(path)
+            with self._lock:
+                with ftp_errors(self, path=path):
+                    cmd = "MDTM " + _encode(_path, self.ftp.encoding)
+                    response = self.ftp.sendcmd(cmd)
+                    mtime = self._parse_ftp_time(response.split()[1])
+                    return epoch_to_datetime(mtime)
+        return super(FTPFS, self).getmodified(path)
+
     def listdir(self, path):
         # type: (Text) -> List[Text]
         _path = self.validatepath(path)

tests/test_ftpfs.py

@@ -261,6 +261,23 @@ def test_getmeta_unicode_path(self):
         del self.fs.features["UTF8"]
         self.assertFalse(self.fs.getmeta().get("unicode_paths"))
 
+    def test_getinfo_modified(self):
+        self.assertIn("MDTM", self.fs.features)
+        self.fs.create("bar")
+        mtime_detail = self.fs.getinfo("bar", ("basic", "details")).modified
+        mtime_modified = self.fs.getmodified("bar")
+        # Microsecond and seconds might not actually be supported by all
+        # FTP commands, so we strip them before comparing if it looks
+        # like at least one of the two values does not contain them.
+        replacement = {}
+        if mtime_detail.microsecond == 0 or mtime_modified.microsecond == 0:
+            replacement["microsecond"] = 0
+        if mtime_detail.second == 0 or mtime_modified.second == 0:
+            replacement["second"] = 0
+        self.assertEqual(
+            mtime_detail.replace(**replacement), mtime_modified.replace(**replacement)
+        )
+
     def test_opener_path(self):
         self.fs.makedir("foo")
         self.fs.writetext("foo/bar", "baz")

tests/test_memoryfs.py

@@ -72,14 +72,13 @@ def test_copy_preserve_time(self):
         self.fs.makedir("bar")
         self.fs.touch("foo/file.txt")
 
-        namespaces = ("details", "modified")
-        src_info = self.fs.getinfo("foo/file.txt", namespaces)
+        src_datetime = self.fs.getmodified("foo/file.txt")
 
         self.fs.copy("foo/file.txt", "bar/file.txt", preserve_time=True)
         self.assertTrue(self.fs.exists("bar/file.txt"))
 
-        dst_info = self.fs.getinfo("bar/file.txt", namespaces)
-        self.assertEqual(dst_info.modified, src_info.modified)
+        dst_datetime = self.fs.getmodified("bar/file.txt")
+        self.assertEqual(dst_datetime, src_datetime)
 
 
 class TestMemoryFile(unittest.TestCase):