Menu

[r29]: / python_webdav / client.py  Maximize  Restore  History

Download this file

152 lines (122 with data), 5.2 kB

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
"""
Client Module
=============
"""
import os
import urllib2
import python_webdav.connection as conn
class Client(object):
""" This is used for accessing a WebDAV service using similar commands
that might be found in a CLI
"""
def __init__(self, webdav_server_uri, webdav_path='.', port=80, realm=''):
"""
The Client module is not yet ready for use. The purpose of this module is to
make WebDAV use easier. The Connection.Client object is for lower level use
while this top level module will hopefully aid in quicker development.
"""
self._connection_settings = dict(host=webdav_server_uri,
path=webdav_path,
port=port, realm=realm)
path = self._connection_settings['path']
if path[-1] != '/' and path != '.':
self._connection_settings['path'] += '/'
self.connection = None
def set_connection(self, username='', password=''):
""" Set up the connection object
:param username: Username if authentication is required for the connection
:type username: String
:param password: Password if authentication is required for the connection
:type password: String
"""
self._connection_settings['username'] = username
self._connection_settings['password'] = password
self.connection = conn.Connection(self._connection_settings)
self.client = conn.Client()
def download_file(self, file_path, dest_path='.'):
""" Download a file from file_path to dest_path
:param file_path: Path to the resource to download
:type file_path: String
:param dest_path: Path to where the downloaded file should be saved
:type dest_path: String
"""
resource_path = urllib2.urlparse.urljoin(self.connection.path,
file_path.strip('/'))
resp, content = self.connection.send_get(resource_path)
file_name = os.path.basename(file_path)
write_to_path = os.path.join(dest_path, file_name)
try:
file_fd = open(write_to_path, 'wb')
file_fd.write(content)
except IOError:
raise
finally:
file_fd.close()
return resp, content
def chdir(self, directory):
""" Change directory from whatever current dir is to directory specified
:param directory: Directory to change to
:type directory: String
"""
# Make sure there's a leading '/'
if not self.connection.path.startswith('/'):
self.connection.path = '/' + self.connection.path
self.connection.path = os.path.realpath(
os.path.join(self.connection.path, directory))
def mkdir(self, path):
""" Make a directory (collection). If path does not start with '/'
it is assumed to be relative to the current working directory.
:param path: Path of the directory to create
:type path: String
"""
if not path.startswith('/'):
path = self.connection.path + '/' + path
self.connection.send_mkcol(path)
# ---------------- UNFINISHED --------------- #
def ls(self, path='', list_format=('F','C','M'), separator='\t', display=True):
"""
:param path: Path of the directory to list
:type path: String
:param list_format: This is the format for the directory listing
The format symbols are:
* T - Type (resourcetype)
* D - Date Time (creationdate)
* F - Filename (href)
* M - Last modified time (getlastmodified)
* A - Attributes (executable)
* E - ETag (getetag)
* C - Content type (getcontenttype)
:type list_format: List
:param separator: Separator to use for formatting output
:type separator: String
:param display: Whether or not to print the output
:type display: Boolean
"""
# Format Map
format_map = {'T': 'resourcetype',
'D': 'creationdate',
'F': 'href',
'M': 'getlastmodified',
'A': 'executable',
'E': 'getetag',
'C': 'getcontenttype'}
# Get the properties for the given path
if not path:
path = self.connection.path
props = self.client.get_properties(self.connection, path)
property_lists = []
for prop in props:
format_string = ''
formatted_list = [getattr(prop, format_map[symbol]) for symbol in list_format]
property_lists.append(formatted_list)
format_string = separator.join(formatted_list)
if display:
print format_string
return property_lists
def pwd(self):
""" Return the working directory
"""
if self.connection:
return self.connection.path
else:
return None