Added support for "last modified time" in getinfo.

A new namespace "modified" was introduced to the `getinfo` function to potentially use cheaper commands compared to the general "details" namespace. In particular, `FTPFS` now uses the MDTM command if supported by the server.

Author: atollk

CHANGELOG.md

@@ -13,6 +13,12 @@ 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).
 
+### 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
 
 - Fixed performance bugs in `fs.copy.copy_dir_if_newer`. Test cases were adapted to catch those bugs in the future.

fs/base.py

@@ -678,6 +678,25 @@ 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.
+
+        """
+        if self.getmeta().get("supports_mtime", False):
+            return self.getinfo(path, namespaces=["modified"]).modified
+        return self.getinfo(path, namespaces=["details"]).modified
+
     def getmeta(self, namespace="standard"):
         # type: (Text) -> Mapping[Text, object]
         """Get meta information regarding a filesystem.
@@ -715,6 +734,8 @@ def getmeta(self, namespace="standard"):
         read_only           `True` if this filesystem is read only.
         supports_rename     `True` if this filesystem supports an
                             `os.rename` operation.
+        supports_mtime      `True` if this filesystem supports a native
+                            operation to retreive the "last modified" time.
         =================== ============================================
 
         Most builtin filesystems will provide all these keys, and third-

fs/copy.py

@@ -398,9 +398,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:
@@ -412,9 +411,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

@@ -15,6 +15,7 @@
 from contextlib import contextmanager
 from ftplib import FTP
 
+
 try:
     from ftplib import FTP_TLS
 except ImportError as err:
@@ -667,6 +668,18 @@ def getinfo(self, path, namespaces=None):
                 }
             )
 
+        if "modified" in namespaces:
+            if "basic" in namespaces or "details" in namespaces:
+                raise ValueError(
+                    'Cannot use the "modified" namespace in combination with others.'
+                )
+            with self._lock:
+                with ftp_errors(self, path=path):
+                    cmd = "MDTM " + _encode(self.validatepath(path), self.ftp.encoding)
+                    response = self.ftp.sendcmd(cmd)
+                modified_info = {"modified": self._parse_ftp_time(response.split()[1])}
+                return Info({"modified": modified_info})
+
         if self.supports_mlst:
             with self._lock:
                 with ftp_errors(self, path=path):
@@ -692,6 +705,7 @@ 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 listdir(self, path):

fs/info.py

@@ -317,9 +317,12 @@ def modified(self):
                 namespace is not in the Info.
 
         """
-        self._require_namespace("details")
-        _time = self._make_datetime(self.get("details", "modified"))
-        return _time
+        try:
+            self._require_namespace("details")
+            return self._make_datetime(self.get("details", "modified"))
+        except MissingInfoNamespace:
+            self._require_namespace("modified")
+            return self._make_datetime(self.get("modified", "modified"))
 
     @property
     def created(self):

fs/osfs.py

@@ -144,6 +144,7 @@ def __init__(
             "network": False,
             "read_only": False,
             "supports_rename": True,
+            "supports_mtime": False,
             "thread_safe": True,
             "unicode_paths": os.path.supports_unicode_filenames,
             "virtual": False,

tests/test_ftpfs.py

@@ -144,7 +144,6 @@ def test_manager_with_host(self):
 @mark.slow
 @unittest.skipIf(platform.python_implementation() == "PyPy", "ftp unreliable with PyPy")
 class TestFTPFS(FSTestCases, unittest.TestCase):
-
     user = "user"
     pasw = "1234"
 
@@ -243,6 +242,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.getinfo("bar", ("modified",)).modified
+        # 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")
@@ -301,7 +317,6 @@ def test_features(self):
 @mark.slow
 @unittest.skipIf(platform.python_implementation() == "PyPy", "ftp unreliable with PyPy")
 class TestAnonFTPFS(FSTestCases, unittest.TestCase):
-
     user = "anonymous"
     pasw = ""
 

tests/test_memoryfs.py

@@ -69,7 +69,6 @@ def test_close_mem_free(self):
 
 
 class TestMemoryFile(unittest.TestCase):
-
     def setUp(self):
         self.fs = memoryfs.MemoryFS()
 

tests/test_opener.py

@@ -300,14 +300,26 @@ def test_user_data_opener(self, app_dir):
     def test_open_ftp(self, mock_FTPFS):
         open_fs("ftp://foo:bar@ftp.example.org")
         mock_FTPFS.assert_called_once_with(
-            "ftp.example.org", passwd="bar", port=21, user="foo", proxy=None, timeout=10, tls=False
+            "ftp.example.org",
+            passwd="bar",
+            port=21,
+            user="foo",
+            proxy=None,
+            timeout=10,
+            tls=False,
         )
 
     @mock.patch("fs.ftpfs.FTPFS")
     def test_open_ftps(self, mock_FTPFS):
         open_fs("ftps://foo:bar@ftp.example.org")
         mock_FTPFS.assert_called_once_with(
-            "ftp.example.org", passwd="bar", port=21, user="foo", proxy=None, timeout=10, tls=True
+            "ftp.example.org",
+            passwd="bar",
+            port=21,
+            user="foo",
+            proxy=None,
+            timeout=10,
+            tls=True,
         )
 
     @mock.patch("fs.ftpfs.FTPFS")

tests/test_wrap.py

@@ -177,7 +177,7 @@ def test_scandir(self):
         ]
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertEqual(sorted(self.cached.scandir("/"), key=key), expected)
-            scandir.assert_has_calls([mock.call('/', namespaces=None, page=None)])
+            scandir.assert_has_calls([mock.call("/", namespaces=None, page=None)])
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertEqual(sorted(self.cached.scandir("/"), key=key), expected)
             scandir.assert_not_called()
@@ -187,7 +187,7 @@ def test_isdir(self):
             self.assertTrue(self.cached.isdir("foo"))
             self.assertFalse(self.cached.isdir("egg"))  # is file
             self.assertFalse(self.cached.isdir("spam"))  # doesn't exist
-            scandir.assert_has_calls([mock.call('/', namespaces=None, page=None)])
+            scandir.assert_has_calls([mock.call("/", namespaces=None, page=None)])
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertTrue(self.cached.isdir("foo"))
             self.assertFalse(self.cached.isdir("egg"))
@@ -199,7 +199,7 @@ def test_isfile(self):
             self.assertTrue(self.cached.isfile("egg"))
             self.assertFalse(self.cached.isfile("foo"))  # is dir
             self.assertFalse(self.cached.isfile("spam"))  # doesn't exist
-            scandir.assert_has_calls([mock.call('/', namespaces=None, page=None)])
+            scandir.assert_has_calls([mock.call("/", namespaces=None, page=None)])
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertTrue(self.cached.isfile("egg"))
             self.assertFalse(self.cached.isfile("foo"))
@@ -211,7 +211,7 @@ def test_getinfo(self):
             self.assertEqual(self.cached.getinfo("foo"), self.fs.getinfo("foo"))
             self.assertEqual(self.cached.getinfo("/"), self.fs.getinfo("/"))
             self.assertNotFound(self.cached.getinfo, "spam")
-            scandir.assert_has_calls([mock.call('/', namespaces=None, page=None)])
+            scandir.assert_has_calls([mock.call("/", namespaces=None, page=None)])
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertEqual(self.cached.getinfo("foo"), self.fs.getinfo("foo"))
             self.assertEqual(self.cached.getinfo("/"), self.fs.getinfo("/"))