Compare commits

...

5 Commits

Author SHA1 Message Date
Ciaran Gultnieks 8159e85172 Allow an auth object to be passed when connecting
This means other authentication methods can be used, e.g. digest
or NTLM, by simply constructing one of the relevant auth objects
supplied with the Requests library.
2012-11-13 08:59:46 +00:00
Ciaran Gultnieks 13c98a464f Support Python 2.5
Seems silly to rule out Python 2.5 (and consequently Debian stable)
for the sake of a few numbers.
2012-11-13 08:55:06 +00:00
Ciaran Gultnieks 350dd9876d Update README re directory listing 2012-11-13 08:54:26 +00:00
Miki Tebeka eb882938d7 adding ls 2012-10-17 11:35:14 -07:00
amnong 8411a9a9a1 Bumped version to 1.0.5 2012-08-25 20:07:42 +03:00
3 changed files with 50 additions and 8 deletions

View File

@ -7,8 +7,7 @@ Features
* Basic authentication
* Creating directories, removing directories and files
* Uploading and downloading files
Note: Directory listing is not supported. I'll implement it upon request.
* Directory listing
Installation
------------

View File

@ -1 +1 @@
__version__ = "1.0.4"
__version__ = "1.0.5"

View File

@ -2,6 +2,10 @@ import requests
import shutil
from numbers import Number
from httplib import responses as HTTP_CODES
from urlparse import urlparse
import xml.etree.cElementTree as xml
from collections import namedtuple
from cStringIO import StringIO
class WebdavException(Exception):
pass
@ -9,12 +13,35 @@ class WebdavException(Exception):
class ConnectionFailed(WebdavException):
pass
def codestr(code):
return HTTP_CODES.get(code, 'UNKNOWN')
File = namedtuple('File', ['name', 'size', 'mtime', 'ctime'])
def prop(elem, name, default=None):
child = elem.find('.//{DAV:}' + name)
return default if child is None else child.text
def elem2file(elem):
return File(
prop(elem, 'href'),
int(prop(elem, 'getcontentlength', 0)),
prop(elem, 'getlastmodified', ''),
prop(elem, 'creationdate', ''),
)
class OperationFailed(WebdavException):
_OPERATIONS = dict(
GET = "download",
PUT = "upload",
DELETE = "delete",
MKCOL = "create directory",
PROPFIND = "list directory",
)
def __init__(self, method, path, expected_code, actual_code):
self.method = method
@ -24,8 +51,8 @@ class OperationFailed(WebdavException):
operation_name = self._OPERATIONS[method]
self.reason = 'Failed to {operation_name} "{path}"'.format(**locals())
expected_codes = (expected_code,) if isinstance(expected_code, Number) else expected_code
expected_codes_str = ", ".join('{} {}'.format(code, HTTP_CODES[code]) for code in expected_codes)
actual_code_str = HTTP_CODES[actual_code]
expected_codes_str = ", ".join('{0} {1}'.format(code, codestr(code)) for code in expected_codes)
actual_code_str = codestr(actual_code)
msg = '''\
{self.reason}.
Operation : {method} {path}
@ -34,11 +61,16 @@ class OperationFailed(WebdavException):
super(OperationFailed, self).__init__(msg)
class Client(object):
def __init__(self, host, port=80, username=None, password=None):
self.baseurl = 'http://{}:{}'.format(host ,port)
def __init__(self, host, port=0, auth=None, username=None, password=None,
protocol='http'):
if not port:
port = 443 if protocol == 'https' else 80
self.baseurl = '{0}://{1}:{2}'.format(protocol, host ,port)
self.cwd = '/'
self.session = requests.session()
if username and password:
if auth:
self.session.auth = auth
elif username and password:
self.session.auth = (username, password)
def _send(self, method, path, expected_code, **kwargs):
url = self._get_url(path)
@ -93,3 +125,14 @@ class Client(object):
with open(local_path, 'wb') as f:
#f.write(response.content)
shutil.copyfileobj(response.raw, f)
def ls(self, remote_path):
headers = {'Depth': '1'}
response = self._send('PROPFIND', remote_path, (207, 301), headers=headers)
# Redirect
if response.status_code == 301:
url = urlparse(response.headers['location'])
return self.ls(url.path)
tree = xml.parse(StringIO(response.content))
return [elem2file(elem) for elem in tree.findall('{DAV:}response')]