use FS URLs in multi and mount fs

Author: willmcgugan

fs/base.py

@@ -1204,7 +1204,7 @@ def check(self):
  Check a filesystem may be used.
 
  Will throw a :class:`~fs.errors.FilesystemClosed` if the
-  filesystem is closed.
+ filesystem is closed.
 
  :returns: None
  :raises fs.errors.FilesystemClosed: if the filesystem

fs/iotools.py

@@ -115,7 +115,6 @@ def __iter__(self):
  return iter(self._f)
 
 
-
 def make_stream(name,
  bin_file,
  mode='r',

fs/mountfs.py

@@ -2,6 +2,8 @@
 from __future__ import print_function
 from __future__ import unicode_literals
 
+from six import text_type
+
 from . import errors
 from .base import FS
 from .memoryfs import MemoryFS
@@ -67,10 +69,19 @@ def mount(self, path, fs):
 
  :param path: A path within the MountFS.
  :type path: str
- :param fs: A filesystem object to mount.
+ :param fs: A filesystem object or FS URL to mount.
  :type fs: :class:`~fs.base.FS`
 
  """
+ if isinstance(fs, text_type):
+ from .opener import open_fs
+ fs = open_fs(fs)
+
+ if not isinstance(fs, FS):
+ raise TypeError(
+ 'fs argument must be an FS object or a FS URL'
+ )
+
  if fs is self:
  raise ValueError('Unable to mount self')
  _path = forcedir(abspath(normpath(path)))

fs/multifs.py

@@ -5,15 +5,21 @@
 from collections import namedtuple
 from operator import itemgetter
 
+from six import text_type
+
 from .base import FS
 from .mode import check_writable
 from . import errors
+from .opener import open_fs
 from .path import abspath, normpath
 
 
 _PrioritizedFS = namedtuple(
  '_PrioritizedFS',
- ['priority', 'fs']
+ [
+ 'priority',
+ 'fs'
+ ]
 )
 
 
@@ -61,8 +67,8 @@ def add_fs(self, name, fs, write=False, priority=0):
  :param name: A unique name to refer to the filesystem being
  added.
  :type name: str
- :param fs: The filesystem to add
- :type fs: Filesystem
+ :param fs: The filesystem to add.
+ :type fs: Filesystem or FS URL.
  :param write: If this value is True, then the ``fs`` will be
  used as the writeable FS.
  :type write: bool
@@ -75,6 +81,14 @@ def add_fs(self, name, fs, write=False, priority=0):
 
  """
 
+ if isinstance(fs, text_type):
+ fs = open_fs(fs)
+
+ if not isinstance(fs, FS):
+ raise TypeError(
+ 'fs argument should be an FS object or FS URL'
+ )
+
  self._filesystems[name] = _PrioritizedFS(
  priority=(priority, self._sort_index),
  fs=fs

fs/opener/appfs.py

@@ -0,0 +1,55 @@
+# coding: utf-8
+"""Defines the MemOpener."""
+
+from __future__ import absolute_import
+from __future__ import print_function
+from __future__ import unicode_literals
+
+from .base import Opener
+from .errors import OpenerError
+from .. import appfs
+
+
+class Opener(Opener):
+
+ protocols = [
+ 'userdata',
+ 'userconf',
+ 'sitedata',
+ 'siteconf',
+ 'usercache',
+ 'userlog'
+ ]
+
+ _protocol_mapping = {
+ 'userdata': appfs.UserDataFS,
+ 'userconf': appfs.UserConfigFS,
+ 'sitedata': appfs.SiteDataFS,
+ 'siteconf': appfs.SiteConfigFS,
+ 'usercache': appfs.UserCacheFS,
+ 'userlog': appfs.UserLogFS
+ }
+
+ def open_fs(self, fs_url, parse_result, writeable, create, cwd):
+ fs_class = self._protocol_mapping[parse_result.protocol]
+
+ tokens = parse_result.resource.split(':', 3)
+ if len(tokens) == 2:
+ appname, author = tokens
+ version = None
+ elif len(tokens) == 3:
+ appname, author, version = tokens
+ else:
+ raise OpenerError(
+ 'resource should be <appname>:<author> '
+ 'or <appname>:<author>:<version'
+ )
+
+ fs_instance = fs_class(
+ appname,
+ author=author,
+ version=version,
+ create=create
+ )
+ return fs_instance
+

fs/wrapfs.py

@@ -70,7 +70,6 @@ def delegate_fs(self):
  """
  return self._wrap_fs
 
-
  def appendbytes(self, path, data):
  self.check()
  _fs, _path = self.delegate_path(path)
@@ -88,7 +87,6 @@ def appendtext(self, path, text,
  errors=errors,
  newline=newline)
 
-
  def getinfo(self, path, namespaces=None):
  self.check()
  _fs, _path = self.delegate_path(path)

setup.py

@@ -47,6 +47,13 @@
  'tar = fs.opener.tarfs:TarOpener',
  'temp = fs.opener.tempfs:TempOpener',
  'zip = fs.opener.zipfs:ZipOpener',
+
+ 'userdata = fs.opener.appfs:Opener',
+ 'userconf = fs.opener.appfs:Opener',
+ 'sitedata = fs.opener.appfs:Opener',
+ 'siteconf = fs.opener.appfs:Opener',
+ 'usercache = fs.opener.appfs:Opener',
+ 'userlog = fs.opener.appfs:Opener',
  ]},
  license="MIT",
  long_description=DESCRIPTION,

tests/test_mountfs.py

@@ -30,15 +30,21 @@ def make_fs(self):
 
 class TestMountFSBehaviours(unittest.TestCase):
 
+ def test_bad_mount(self):
+ mount_fs = MountFS()
+ with self.assertRaises(TypeError):
+ mount_fs.mount('foo', 5)
+ with self.assertRaises(TypeError):
+ mount_fs.mount('foo', b'bar')
+
  def test_listdir(self):
  mount_fs = MountFS()
  self.assertEqual(mount_fs.listdir('/'), [])
  m1 = MemoryFS()
- m2 = TempFS()
  m3 = MemoryFS()
  m4 = TempFS()
  mount_fs.mount('/m1', m1)
- mount_fs.mount('/m2', m2)
+ mount_fs.mount('/m2', 'temp://')
  mount_fs.mount('/m3', m3)
  with self.assertRaises(MountError):
  mount_fs.mount('/m3/foo', m4)

tests/test_multifs.py

@@ -2,6 +2,8 @@
 
 import unittest
 
+from six import text_type
+
 from fs.multifs import MultiFS
 from fs.memoryfs import MemoryFS
 from fs import errors
@@ -64,6 +66,18 @@ def test_no_auto_close(self):
  self.assertFalse(m1.isclosed())
  self.assertFalse(m2.isclosed())
 
+ def test_opener(self):
+ """Test use of FS URLs."""
+ multi_fs = MultiFS()
+ with self.assertRaises(TypeError):
+ multi_fs.add_fs(u'foo', 5)
+ multi_fs.add_fs(u'f1', u'mem://')
+ multi_fs.add_fs(u'f2', u'temp://')
+ self.assertIsInstance(
+ multi_fs.get_fs(u'f1'),
+ MemoryFS
+ )
+
  def test_priority(self):
  """Test priority order is working"""
  m1 = MemoryFS()

tests/test_opener.py

@@ -202,3 +202,24 @@ def test_open_fs(self):
  mem_fs = opener.open_fs("mem://")
  mem_fs_2 = opener.open_fs(mem_fs)
  self.assertEqual(mem_fs, mem_fs_2)
+
+ def test_open_userdata(self):
+ with self.assertRaises(errors.OpenerError):
+ opener.open_fs('userdata://foo:bar:baz:egg')
+
+ app_fs = opener.open_fs(
+ 'userdata://fstest:willmcgugan:1.0',
+ create=True
+ )
+ self.assertEqual(app_fs.app_dirs.appname, 'fstest')
+ self.assertEqual(app_fs.app_dirs.appauthor, 'willmcgugan')
+ self.assertEqual(app_fs.app_dirs.version, '1.0')
+
+ def test_open_userdata_no_version(self):
+ app_fs = opener.open_fs(
+ 'userdata://fstest:willmcgugan',
+ create=True
+ )
+ self.assertEqual(app_fs.app_dirs.appname, 'fstest')
+ self.assertEqual(app_fs.app_dirs.appauthor, 'willmcgugan')
+ self.assertEqual(app_fs.app_dirs.version, None)