forked from algorithmiaio/algorithmia-python
-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathdatadirectory.py
More file actions
144 lines (121 loc) · 4.9 KB
/
Copy pathdatadirectory.py
File metadata and controls
144 lines (121 loc) · 4.9 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
'Algorithmia Data API Client (python)'
import json
import re
import six
import tempfile
import Algorithmia
from Algorithmia.datafile import DataFile
from Algorithmia.data import DataObject, DataObjectType
from Algorithmia.errors import DataApiError
from Algorithmia.util import getParentAndBase, pathJoin
from Algorithmia.acl import Acl
class DataDirectory(DataObject):
def __init__(self, client, dataUrl):
super(DataDirectory, self).__init__(DataObjectType.directory)
self.client = client
# Parse dataUrl
self.path = re.sub(r'^data://|^/', '', dataUrl)
self.url = DataDirectory._getUrl(self.path)
@staticmethod
def _getUrl(path):
return '/v1/data/' + path
def set_attributes(self, response_json):
# Nothing to set for now
pass
def getName(self):
_, name = getParentAndBase(self.path)
return name
def exists(self):
# Heading a directory apparently isn't a valid operation
response = self.client.getHelper(self.url)
return (response.status_code == 200)
def create(self, acl=None):
'''Creates a directory, optionally include Acl argument to set permissions'''
parent, name = getParentAndBase(self.path)
json = { 'name': name }
if acl is not None:
json['acl'] = acl.to_api_param()
response = self.client.postJsonHelper(DataDirectory._getUrl(parent), json, False)
if (response.status_code != 200):
raise DataApiError("Directory creation failed: " + str(response.content))
def delete(self, force=False):
# Delete from data api
url = self.url
if force:
url += '?force=true'
result = self.client.deleteHelper(url)
if 'error' in result:
raise DataApiError(result['error']['message'])
else:
return True
def file(self, name):
return DataFile(self.client, pathJoin(self.path, name))
def files(self):
return self._get_directory_iterator(DataObjectType.file)
def dir(self, name):
return DataDirectory(self.client, pathJoin(self.path, name))
def dirs(self):
return self._get_directory_iterator(DataObjectType.directory)
def list(self):
return self._get_directory_iterator()
def get_permissions(self):
'''
Returns permissions for this directory or None if it's a special collection such as
.session or .algo
'''
response = self.client.getHelper(self.url, acl='true')
if response.status_code != 200:
raise DataApiError('Unable to get permissions:' + str(response.content))
content = response.json()
if 'acl' in content:
return Acl.from_acl_response(content['acl'])
else:
return None
def update_permissions(self, acl):
params = {'acl':acl.to_api_param()}
response = self.client.patchHelper(self.url, params)
if response.status_code != 200:
raise DataApiError('Unable to update permissions: ' + response.json()['error']['message'])
return True
def _get_directory_iterator(self, type_filter=None):
marker = None
first = True
while first or (marker is not None and len(marker) > 0):
first = False
url = self.url
query_params= {}
if marker:
query_params['marker'] = marker
response = self.client.getHelper(url, **query_params)
if response.status_code != 200:
raise DataApiError("Directory iteration failed: " + str(response.content))
responseContent = response.content
if isinstance(responseContent, six.binary_type):
responseContent = responseContent.decode()
content = json.loads(responseContent)
if 'marker' in content:
marker = content['marker']
else:
marker = None
if type_filter is DataObjectType.directory or type_filter is None:
for d in self._iterate_directories(content):
yield d
if type_filter is DataObjectType.file or type_filter is None:
for f in self._iterate_files(content):
yield f
def _iterate_directories(self, content):
directories = []
if 'folders' in content:
for dir_info in content['folders']:
d = DataDirectory(self.client, pathJoin(self.path, dir_info['name']))
d.set_attributes(dir_info)
directories.append(d)
return directories
def _iterate_files(self, content):
files = []
if 'files' in content:
for file_info in content['files']:
f = DataFile(self.client, pathJoin(self.path, file_info['filename']))
f.set_attributes(file_info)
files.append(f)
return files