Skip to content

P
pyfs
Commit 081673c0 by willmcgugan
Options

Fixes for Py2.5 and Py2.7

parent ba79c2c2
Showing with 100 additions and 20 deletions
fs/expose/xmlrpc.py
...@@ -17,7 +17,7 @@ an FS object, which can then be exposed using whatever server you choose ...@@ -17,7 +17,7 @@ an FS object, which can then be exposed using whatever server you choose
import xmlrpclib import xmlrpclib
from SimpleXMLRPCServer import SimpleXMLRPCServer from SimpleXMLRPCServer import SimpleXMLRPCServer
from datetime import datetime
class RPCFSInterface(object): class RPCFSInterface(object):
"""Wrapper to expose an FS via a XML-RPC compatible interface. """Wrapper to expose an FS via a XML-RPC compatible interface.
...@@ -27,6 +27,7 @@ class RPCFSInterface(object): ...@@ -27,6 +27,7 @@ class RPCFSInterface(object):
""" """
def __init__(self, fs): def __init__(self, fs):
super(RPCFSInterface, self).__init__()
self.fs = fs self.fs = fs
def encode_path(self, path): def encode_path(self, path):
...@@ -98,6 +99,10 @@ class RPCFSInterface(object): ...@@ -98,6 +99,10 @@ class RPCFSInterface(object):
def settimes(self, path, accessed_time, modified_time): def settimes(self, path, accessed_time, modified_time):
path = self.decode_path(path) path = self.decode_path(path)
if isinstance(accessed_time, xmlrpclib.DateTime):
accessed_time = datetime.strptime(accessed_time.value, "%Y%m%dT%H:%M:%S")
if isinstance(modified_time, xmlrpclib.DateTime):
modified_time = datetime.strptime(modified_time.value, "%Y%m%dT%H:%M:%S")
return self.fs.settimes(path, accessed_time, modified_time) return self.fs.settimes(path, accessed_time, modified_time)
def getinfo(self, path): def getinfo(self, path):
......
fs/rpcfs.py
...@@ -10,6 +10,7 @@ class from the :mod:`fs.expose.xmlrpc` module. ...@@ -10,6 +10,7 @@ class from the :mod:`fs.expose.xmlrpc` module.
import xmlrpclib import xmlrpclib
import socket import socket
import threading
from fs.base import * from fs.base import *
from fs.errors import * from fs.errors import *
...@@ -106,14 +107,15 @@ class RPCFS(FS): ...@@ -106,14 +107,15 @@ class RPCFS(FS):
:param uri: address of the server :param uri: address of the server
""" """
super(RPCFS, self).__init__(thread_synchronize=True)
self.uri = uri self.uri = uri
self._transport = transport self._transport = transport
self.proxy = self._make_proxy() self.proxy = self._make_proxy()
FS.__init__(self,thread_synchronize=False)
self.isdir('/') self.isdir('/')
@synchronize
def _make_proxy(self): def _make_proxy(self):
kwds = dict(allow_none=True) kwds = dict(allow_none=True, use_datetime=True)
if self._transport is not None: if self._transport is not None:
proxy = xmlrpclib.ServerProxy(self.uri,self._transport,**kwds) proxy = xmlrpclib.ServerProxy(self.uri,self._transport,**kwds)
...@@ -125,6 +127,7 @@ class RPCFS(FS): ...@@ -125,6 +127,7 @@ class RPCFS(FS):
def __str__(self): def __str__(self):
return '<RPCFS: %s>' % (self.uri,) return '<RPCFS: %s>' % (self.uri,)
@synchronize
def __getstate__(self): def __getstate__(self):
state = super(RPCFS,self).__getstate__() state = super(RPCFS,self).__getstate__()
try: try:
...@@ -132,11 +135,12 @@ class RPCFS(FS): ...@@ -132,11 +135,12 @@ class RPCFS(FS):
except KeyError: except KeyError:
pass pass
return state return state
def __setstate__(self, state): def __setstate__(self, state):
for (k,v) in state.iteritems(): for (k,v) in state.iteritems():
self.__dict__[k] = v self.__dict__[k] = v
self.proxy = self._make_proxy() self._lock = threading.RLock()
self.proxy = self._make_proxy()
def encode_path(self, path): def encode_path(self, path):
"""Encode a filesystem path for sending over the wire. """Encode a filesystem path for sending over the wire.
...@@ -151,14 +155,18 @@ class RPCFS(FS): ...@@ -151,14 +155,18 @@ class RPCFS(FS):
"""Decode paths arriving over the wire.""" """Decode paths arriving over the wire."""
return path.decode("base64").decode("utf8") return path.decode("base64").decode("utf8")
@synchronize
def getmeta(self, meta_name, default=NoDefaultMeta): def getmeta(self, meta_name, default=NoDefaultMeta):
if default is NoDefaultMeta: if default is NoDefaultMeta:
return self.proxy.getmeta(meta_name) return self.proxy.getmeta(meta_name)
else: else:
return self.proxy.getmeta_default(meta_name, default) return self.proxy.getmeta_default(meta_name, default)
@synchronize
def hasmeta(self, meta_name): def hasmeta(self, meta_name):
return self.proxy.hasmeta(meta_name) return self.proxy.hasmeta(meta_name)
@synchronize
def open(self, path, mode="r"): def open(self, path, mode="r"):
# TODO: chunked transport of large files # TODO: chunked transport of large files
path = self.encode_path(path) path = self.encode_path(path)
...@@ -182,33 +190,50 @@ class RPCFS(FS): ...@@ -182,33 +190,50 @@ class RPCFS(FS):
f.seek(0,2) f.seek(0,2)
oldflush = f.flush oldflush = f.flush
oldclose = f.close oldclose = f.close
oldtruncate = f.truncate oldtruncate = f.truncate
def newflush(): def newflush():
oldflush() self._lock.acquire()
self.proxy.set_contents(path,xmlrpclib.Binary(f.getvalue())) try:
oldflush()
self.proxy.set_contents(path,xmlrpclib.Binary(f.getvalue()))
finally:
self._lock.release()
def newclose(): def newclose():
f.flush() self._lock.acquire()
oldclose() try:
f.flush()
oldclose()
finally:
self._lock.release()
def newtruncate(size=None): def newtruncate(size=None):
oldtruncate(size) self._lock.acquire()
f.flush() try:
oldtruncate(size)
f.flush()
finally:
self._lock.release()
f.flush = newflush f.flush = newflush
f.close = newclose f.close = newclose
f.truncate = newtruncate f.truncate = newtruncate
return f return f
@synchronize
def exists(self, path): def exists(self, path):
path = self.encode_path(path) path = self.encode_path(path)
return self.proxy.exists(path) return self.proxy.exists(path)
@synchronize
def isdir(self, path): def isdir(self, path):
path = self.encode_path(path) path = self.encode_path(path)
return self.proxy.isdir(path) return self.proxy.isdir(path)
@synchronize
def isfile(self, path): def isfile(self, path):
path = self.encode_path(path) path = self.encode_path(path)
return self.proxy.isfile(path) return self.proxy.isfile(path)
@synchronize
def listdir(self, path="./", wildcard=None, full=False, absolute=False, dirs_only=False, files_only=False): def listdir(self, path="./", wildcard=None, full=False, absolute=False, dirs_only=False, files_only=False):
enc_path = self.encode_path(path) enc_path = self.encode_path(path)
if not callable(wildcard): if not callable(wildcard):
...@@ -226,69 +251,84 @@ class RPCFS(FS): ...@@ -226,69 +251,84 @@ class RPCFS(FS):
entries = [abspath(pathjoin(path,e)) for e in entries] entries = [abspath(pathjoin(path,e)) for e in entries]
return entries return entries
@synchronize
def makedir(self, path, recursive=False, allow_recreate=False): def makedir(self, path, recursive=False, allow_recreate=False):
path = self.encode_path(path) path = self.encode_path(path)
return self.proxy.makedir(path,recursive,allow_recreate) return self.proxy.makedir(path,recursive,allow_recreate)
@synchronize
def remove(self, path): def remove(self, path):
path = self.encode_path(path) path = self.encode_path(path)
return self.proxy.remove(path) return self.proxy.remove(path)
@synchronize
def removedir(self, path, recursive=False, force=False): def removedir(self, path, recursive=False, force=False):
path = self.encode_path(path) path = self.encode_path(path)
return self.proxy.removedir(path,recursive,force) return self.proxy.removedir(path,recursive,force)
@synchronize
def rename(self, src, dst): def rename(self, src, dst):
src = self.encode_path(src) src = self.encode_path(src)
dst = self.encode_path(dst) dst = self.encode_path(dst)
return self.proxy.rename(src,dst) return self.proxy.rename(src,dst)
@synchronize
def settimes(self, path, accessed_time, modified_time): def settimes(self, path, accessed_time, modified_time):
path = self.encode_path(path) path = self.encode_path(path)
return self.proxy.settimes(path, accessed_time, modified_time) return self.proxy.settimes(path, accessed_time, modified_time)
@synchronize
def getinfo(self, path): def getinfo(self, path):
path = self.encode_path(path) path = self.encode_path(path)
return self.proxy.getinfo(path) return self.proxy.getinfo(path)
@synchronize
def desc(self, path): def desc(self, path):
path = self.encode_path(path) path = self.encode_path(path)
return self.proxy.desc(path) return self.proxy.desc(path)
@synchronize
def getxattr(self, path, attr, default=None): def getxattr(self, path, attr, default=None):
path = self.encode_path(path) path = self.encode_path(path)
attr = self.encode_path(attr) attr = self.encode_path(attr)
return self.fs.getxattr(path,attr,default) return self.fs.getxattr(path,attr,default)
@synchronize
def setxattr(self, path, attr, value): def setxattr(self, path, attr, value):
path = self.encode_path(path) path = self.encode_path(path)
attr = self.encode_path(attr) attr = self.encode_path(attr)
return self.fs.setxattr(path,attr,value) return self.fs.setxattr(path,attr,value)
@synchronize
def delxattr(self, path, attr): def delxattr(self, path, attr):
path = self.encode_path(path) path = self.encode_path(path)
attr = self.encode_path(attr) attr = self.encode_path(attr)
return self.fs.delxattr(path,attr) return self.fs.delxattr(path,attr)
@synchronize
def listxattrs(self, path): def listxattrs(self, path):
path = self.encode_path(path) path = self.encode_path(path)
return [self.decode_path(a) for a in self.fs.listxattrs(path)] return [self.decode_path(a) for a in self.fs.listxattrs(path)]
@synchronize
def copy(self, src, dst, overwrite=False, chunk_size=16384): def copy(self, src, dst, overwrite=False, chunk_size=16384):
src = self.encode_path(src) src = self.encode_path(src)
dst = self.encode_path(dst) dst = self.encode_path(dst)
return self.proxy.copy(src,dst,overwrite,chunk_size) return self.proxy.copy(src,dst,overwrite,chunk_size)
@synchronize
def move(self, src, dst, overwrite=False, chunk_size=16384): def move(self, src, dst, overwrite=False, chunk_size=16384):
src = self.encode_path(src) src = self.encode_path(src)
dst = self.encode_path(dst) dst = self.encode_path(dst)
return self.proxy.move(src,dst,overwrite,chunk_size) return self.proxy.move(src,dst,overwrite,chunk_size)
@synchronize
def movedir(self, src, dst, overwrite=False, ignore_errors=False, chunk_size=16384): def movedir(self, src, dst, overwrite=False, ignore_errors=False, chunk_size=16384):
src = self.encode_path(src) src = self.encode_path(src)
dst = self.encode_path(dst) dst = self.encode_path(dst)
return self.proxy.movedir(src, dst, overwrite, ignore_errors, chunk_size) return self.proxy.movedir(src, dst, overwrite, ignore_errors, chunk_size)
@synchronize
def copydir(self, src, dst, overwrite=False, ignore_errors=False, chunk_size=16384): def copydir(self, src, dst, overwrite=False, ignore_errors=False, chunk_size=16384):
src = self.encode_path(src) src = self.encode_path(src)
dst = self.encode_path(dst) dst = self.encode_path(dst)
......
fs/sftpfs.py
...@@ -198,16 +198,18 @@ class SFTPFS(FS): ...@@ -198,16 +198,18 @@ class SFTPFS(FS):
def __del__(self): def __del__(self):
self.close() self.close()
@synchronize
def __getstate__(self): def __getstate__(self):
state = super(SFTPFS,self).__getstate__() state = super(SFTPFS,self).__getstate__()
del state["_tlocal"] del state["_tlocal"]
if self._owns_transport: if self._owns_transport:
state['_transport'] = self._transport.getpeername() state['_transport'] = self._transport.getpeername()
return state return state
def __setstate__(self,state): def __setstate__(self,state):
for (k,v) in state.iteritems(): for (k,v) in state.iteritems():
self.__dict__[k] = v self.__dict__[k] = v
self._lock = threading.RLock()
self._tlocal = thread_local() self._tlocal = thread_local()
if self._owns_transport: if self._owns_transport:
self._transport = paramiko.Transport(self._transport) self._transport = paramiko.Transport(self._transport)
...@@ -218,12 +220,13 @@ class SFTPFS(FS): ...@@ -218,12 +220,13 @@ class SFTPFS(FS):
try: try:
return self._tlocal.client return self._tlocal.client
except AttributeError: except AttributeError:
if self._transport is None: #if self._transport is None:
return self._client # return self._client
client = paramiko.SFTPClient.from_transport(self._transport) client = paramiko.SFTPClient.from_transport(self._transport)
self._tlocal.client = client self._tlocal.client = client
return client return client
@synchronize
def close(self): def close(self):
"""Close the connection to the remote server.""" """Close the connection to the remote server."""
if not self.closed: if not self.closed:
...@@ -256,6 +259,7 @@ class SFTPFS(FS): ...@@ -256,6 +259,7 @@ class SFTPFS(FS):
url = 'sftp://%s%s' % (self.hostname.rstrip('/'), abspath(path)) url = 'sftp://%s%s' % (self.hostname.rstrip('/'), abspath(path))
return url return url
@synchronize
@convert_os_errors @convert_os_errors
def open(self,path,mode="rb",bufsize=-1): def open(self,path,mode="rb",bufsize=-1):
npath = self._normpath(path) npath = self._normpath(path)
...@@ -275,6 +279,7 @@ class SFTPFS(FS): ...@@ -275,6 +279,7 @@ class SFTPFS(FS):
f.truncate = new_truncate f.truncate = new_truncate
return f return f
@synchronize
def desc(self, path): def desc(self, path):
npath = self._normpath(path) npath = self._normpath(path)
if self.hostname: if self.hostname:
...@@ -283,6 +288,7 @@ class SFTPFS(FS): ...@@ -283,6 +288,7 @@ class SFTPFS(FS):
addr, port = self._transport.getpeername() addr, port = self._transport.getpeername()
return u'sftp://%s:%i%s' % (addr, port, self.client.normalize(npath)) return u'sftp://%s:%i%s' % (addr, port, self.client.normalize(npath))
@synchronize
@convert_os_errors @convert_os_errors
def exists(self,path): def exists(self,path):
if path in ('', '/'): if path in ('', '/'):
...@@ -296,6 +302,7 @@ class SFTPFS(FS): ...@@ -296,6 +302,7 @@ class SFTPFS(FS):
raise raise
return True return True
@synchronize
@convert_os_errors @convert_os_errors
def isdir(self,path): def isdir(self,path):
if path in ('', '/'): if path in ('', '/'):
...@@ -309,6 +316,7 @@ class SFTPFS(FS): ...@@ -309,6 +316,7 @@ class SFTPFS(FS):
raise raise
return statinfo.S_ISDIR(stat.st_mode) return statinfo.S_ISDIR(stat.st_mode)
@synchronize
@convert_os_errors @convert_os_errors
def isfile(self,path): def isfile(self,path):
npath = self._normpath(path) npath = self._normpath(path)
...@@ -320,6 +328,7 @@ class SFTPFS(FS): ...@@ -320,6 +328,7 @@ class SFTPFS(FS):
raise raise
return statinfo.S_ISREG(stat.st_mode) return statinfo.S_ISREG(stat.st_mode)
@synchronize
@convert_os_errors @convert_os_errors
def listdir(self,path="./",wildcard=None,full=False,absolute=False,dirs_only=False,files_only=False): def listdir(self,path="./",wildcard=None,full=False,absolute=False,dirs_only=False,files_only=False):
npath = self._normpath(path) npath = self._normpath(path)
...@@ -360,7 +369,7 @@ class SFTPFS(FS): ...@@ -360,7 +369,7 @@ class SFTPFS(FS):
return self._listdir_helper(path, paths, wildcard, full, absolute, False, False) return self._listdir_helper(path, paths, wildcard, full, absolute, False, False)
@synchronize
@convert_os_errors @convert_os_errors
def listdirinfo(self,path="./",wildcard=None,full=False,absolute=False,dirs_only=False,files_only=False): def listdirinfo(self,path="./",wildcard=None,full=False,absolute=False,dirs_only=False,files_only=False):
npath = self._normpath(path) npath = self._normpath(path)
...@@ -404,12 +413,13 @@ class SFTPFS(FS): ...@@ -404,12 +413,13 @@ class SFTPFS(FS):
return [(p, getinfo(p)) for p in return [(p, getinfo(p)) for p in
self._listdir_helper(path, paths, wildcard, full, absolute, False, False)] self._listdir_helper(path, paths, wildcard, full, absolute, False, False)]
@synchronize
@convert_os_errors @convert_os_errors
def makedir(self,path,recursive=False,allow_recreate=False): def makedir(self,path,recursive=False,allow_recreate=False):
npath = self._normpath(path) npath = self._normpath(path)
try: try:
self.client.mkdir(npath) self.client.mkdir(npath)
except IOError, e: except IOError, _e:
# Error code is unreliable, try to figure out what went wrong # Error code is unreliable, try to figure out what went wrong
try: try:
stat = self.client.stat(npath) stat = self.client.stat(npath)
...@@ -431,6 +441,7 @@ class SFTPFS(FS): ...@@ -431,6 +441,7 @@ class SFTPFS(FS):
else: else:
raise ResourceInvalidError(path,msg="Can't create directory, there's already a file of that name: %(path)s") raise ResourceInvalidError(path,msg="Can't create directory, there's already a file of that name: %(path)s")
@synchronize
@convert_os_errors @convert_os_errors
def remove(self,path): def remove(self,path):
npath = self._normpath(path) npath = self._normpath(path)
...@@ -443,6 +454,7 @@ class SFTPFS(FS): ...@@ -443,6 +454,7 @@ class SFTPFS(FS):
raise ResourceInvalidError(path,msg="Cannot use remove() on a directory: %(path)s") raise ResourceInvalidError(path,msg="Cannot use remove() on a directory: %(path)s")
raise raise
@synchronize
@convert_os_errors @convert_os_errors
def removedir(self,path,recursive=False,force=False): def removedir(self,path,recursive=False,force=False):
npath = self._normpath(path) npath = self._normpath(path)
...@@ -473,6 +485,7 @@ class SFTPFS(FS): ...@@ -473,6 +485,7 @@ class SFTPFS(FS):
except DirectoryNotEmptyError: except DirectoryNotEmptyError:
pass pass
@synchronize
@convert_os_errors @convert_os_errors
def rename(self,src,dst): def rename(self,src,dst):
nsrc = self._normpath(src) nsrc = self._normpath(src)
...@@ -486,6 +499,7 @@ class SFTPFS(FS): ...@@ -486,6 +499,7 @@ class SFTPFS(FS):
raise ParentDirectoryMissingError(dst) raise ParentDirectoryMissingError(dst)
raise raise
@synchronize
@convert_os_errors @convert_os_errors
def move(self,src,dst,overwrite=False,chunk_size=16384): def move(self,src,dst,overwrite=False,chunk_size=16384):
nsrc = self._normpath(src) nsrc = self._normpath(src)
...@@ -503,6 +517,7 @@ class SFTPFS(FS): ...@@ -503,6 +517,7 @@ class SFTPFS(FS):
raise ParentDirectoryMissingError(dst,msg="Destination directory does not exist: %(path)s") raise ParentDirectoryMissingError(dst,msg="Destination directory does not exist: %(path)s")
raise raise
@synchronize
@convert_os_errors @convert_os_errors
def movedir(self,src,dst,overwrite=False,ignore_errors=False,chunk_size=16384): def movedir(self,src,dst,overwrite=False,ignore_errors=False,chunk_size=16384):
nsrc = self._normpath(src) nsrc = self._normpath(src)
...@@ -537,6 +552,7 @@ class SFTPFS(FS): ...@@ -537,6 +552,7 @@ class SFTPFS(FS):
info['modified_time'] = fromtimestamp(mt) info['modified_time'] = fromtimestamp(mt)
return info return info
@synchronize
@convert_os_errors @convert_os_errors
def getinfo(self, path): def getinfo(self, path):
npath = self._normpath(path) npath = self._normpath(path)
...@@ -554,6 +570,7 @@ class SFTPFS(FS): ...@@ -554,6 +570,7 @@ class SFTPFS(FS):
info['modified_time'] = datetime.datetime.fromtimestamp(mt) info['modified_time'] = datetime.datetime.fromtimestamp(mt)
return info return info
@synchronize
@convert_os_errors @convert_os_errors
def getsize(self, path): def getsize(self, path):
npath = self._normpath(path) npath = self._normpath(path)
......
tox.ini
[tox] [tox]
envlist = py25,py26,py27 envlist = py25,py26,py27
[testenv] [testenv]
commands = nosetests deps=dexml
paramiko
boto
nose
mako
pyftpdlib
[testenv:py25]
commands = nosetests -v \
[]
[testenv:py26]
commands = nosetests -v \
[]
[testenv:py27]
commands = nosetests -v \
[]
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment