1# http://pyrocko.org - GPLv3
2#
3# The Pyrocko Developers, 21st Century
4# ---|P------/S----------~Lg----------
7import re
9from pyrocko import guts
10from pyrocko.guts import make_typed_list_class, String, StringChoice, List, \
11 Int, Object, Union, Bool, Defer
13from pyrocko.io.io_common import FileLoadError
15ResourceTypeList = make_typed_list_class(String)
18guts_prefix = 'wadl'
19guts_xmlns = 'http://wadl.dev.java.net/2009/02'
21re_rmsite = re.compile(r'https?://[^/]+')
22re_multisl = re.compile(r'/+')
25def clean_path(p):
26 p = re_rmsite.sub('', p)
27 p = re_multisl.sub('/', p)
28 if not p.startswith('/'):
29 p = '/' + p
31 return p
34class HTTPMethods(StringChoice):
35 choices = [
36 'GET',
37 'POST',
38 'PUT',
39 'HEAD',
40 'DELETE']
43UriList = make_typed_list_class(String)
45StatusCodeList = make_typed_list_class(Int)
48class ParamStyle(StringChoice):
49 choices = [
50 'plain',
51 'query',
52 'matrix',
53 'header',
54 'template']
57class Doc(Object):
58 xmltagname = 'doc'
59 title = String.T(optional=True, xmlstyle='attribute')
62class Method2(Union):
63 members = [HTTPMethods, String]
66class Include(Object):
67 xmltagname = 'include'
68 href = String.T(optional=True, xmlstyle='attribute')
69 doc_list = List.T(Doc.T())
72class Option(Object):
73 xmltagname = 'option'
74 value = String.T(xmlstyle='attribute')
75 media_type = String.T(optional=True, xmlstyle='attribute')
76 doc_list = List.T(Doc.T())
79class Link(Object):
80 xmltagname = 'link'
81 resource_type = String.T(
82 optional=True, xmlstyle='attribute', xmltagname='resource_type')
83 rel = String.T(optional=True, xmlstyle='attribute')
84 rev = String.T(optional=True, xmlstyle='attribute')
85 doc_list = List.T(Doc.T())
88class Grammars(Object):
89 xmltagname = 'grammars'
90 doc_list = List.T(Doc.T())
91 include_list = List.T(Include.T())
94class DerefError(Exception):
95 pass
98class Element(Object):
100 def __init__(self, *args, **kwargs):
101 Object.__init__(self, *args, **kwargs)
102 self._hyper = None
104 def _update(self, hyper):
105 self._hyper = hyper
107 id_ = getattr(self, 'id', None)
109 if id_:
110 hyper[self.xmltagname][id_] = self
112 for child in self.get_children():
113 child._update(hyper)
115 def get_children(self):
116 raise NotImplementedError()
118 def deref(self):
119 if not self._hyper:
120 raise Exception('Must call _update() before calling deref()')
122 obj = self
123 seen = set()
124 while True:
125 seen.add(obj)
126 href = getattr(obj, 'href', None)
127 if href:
128 try:
129 obj = obj._hyper[obj.xmltagname][href.lstrip('#')]
130 if obj in seen:
131 raise DerefError('cyclic reference')
133 except KeyError:
134 raise DerefError(href)
136 else:
137 return obj
140class Param(Element):
141 xmltagname = 'param'
142 href = String.T(optional=True, xmlstyle='attribute')
143 name = String.T(optional=True, xmlstyle='attribute')
144 style = ParamStyle.T(optional=True, xmlstyle='attribute')
145 id = String.T(optional=True, xmlstyle='attribute')
146 type = String.T(default='xs:string', optional=True, xmlstyle='attribute')
147 default = String.T(optional=True, xmlstyle='attribute')
148 required = Bool.T(default='false', optional=True, xmlstyle='attribute')
149 repeating = Bool.T(default='false', optional=True, xmlstyle='attribute')
150 fixed = String.T(optional=True, xmlstyle='attribute')
151 path = String.T(optional=True, xmlstyle='attribute')
152 doc_list = List.T(Doc.T())
153 option_list = List.T(Option.T())
154 link = Link.T(optional=True)
156 def describe(self, indent):
157 return indent + self.name
159 def get_children(self):
160 return []
163class Representation(Element):
164 xmltagname = 'representation'
165 id = String.T(optional=True, xmlstyle='attribute')
166 element = String.T(optional=True, xmlstyle='attribute')
167 media_type = String.T(optional=True, xmlstyle='attribute')
168 href = String.T(optional=True, xmlstyle='attribute')
169 profile = UriList.T(optional=True, xmlstyle='attribute')
170 doc_list = List.T(Doc.T())
171 param_list = List.T(Param.T())
173 def get_children(self):
174 return self.param_list
177class Request(Element):
178 xmltagname = 'request'
179 doc_list = List.T(Doc.T())
180 param_list = List.T(Param.T())
181 representation_list = List.T(Representation.T())
183 def iter_params(self):
184 for param in self.param_list:
185 param = param.deref()
186 yield param
188 def describe(self, indent):
189 lines = []
190 for param in self.iter_params():
191 lines.append(param.describe(indent))
193 return lines
195 def get_children(self):
196 return self.param_list + self.representation_list
199class Response(Element):
200 xmltagname = 'response'
201 status = StatusCodeList.T(optional=True, xmlstyle='attribute')
202 doc_list = List.T(Doc.T())
203 param_list = List.T(Param.T())
204 representation_list = List.T(Representation.T())
206 def get_children(self):
207 return self.param_list + self.representation_list
210class Method(Element):
211 xmltagname = 'method'
212 id = String.T(optional=True, xmlstyle='attribute')
213 name = String.T(optional=True, xmlstyle='attribute')
214 href = String.T(optional=True, xmlstyle='attribute')
215 doc_list = List.T(Doc.T())
216 request = Request.T(optional=True)
217 response_list = List.T(Response.T())
219 def describe(self, indent):
220 lines = [indent + self.name]
221 if self.request:
222 lines.extend(self.request.describe(' ' + indent))
224 return lines
226 def get_children(self):
227 return ([self.request] if self.request else []) + self.response_list
230class Resource(Element):
231 xmltagname = 'resource'
232 id = String.T(optional=True, xmlstyle='attribute')
233 type = String.T(optional=True, xmlstyle='attribute')
234 query_type = String.T(
235 default='application/x-www-form-urlencoded',
236 optional=True,
237 xmlstyle='attribute')
238 path = String.T(optional=True, xmlstyle='attribute')
239 doc_list = List.T(Doc.T())
240 param_list = List.T(Param.T())
241 method_list = List.T(Method.T())
242 resource_list = List.T(Defer('Resource.T'))
244 def iter_resources(self):
245 yield self.path, self
246 for res in self.resource_list:
247 yield self.path + '/' + res.path, res
249 def iter_methods(self):
250 for method in self.method_list:
251 method = method.deref()
252 yield method
254 def describe(self, indent):
255 lines = []
256 for met in self.iter_methods():
257 lines.extend(met.describe(' ' + indent))
259 return lines
261 def get_children(self):
262 return self.param_list + self.method_list + self.resource_list
265class Resources(Element):
266 xmltagname = 'resources'
267 base = String.T(optional=True, xmlstyle='attribute')
268 doc_list = List.T(Doc.T())
269 resource_list = List.T(Resource.T())
271 def iter_resources(self):
272 for res in self.resource_list:
273 for p, sr in res.iter_resources():
274 yield self.base + '/' + p, sr
276 def get_children(self):
277 return self.resource_list
280class ResourceType(Element):
281 xmltagname = 'resource_type'
282 id = String.T(optional=True, xmlstyle='attribute')
283 doc_list = List.T(Doc.T())
284 param_list = List.T(Param.T())
285 method_list = List.T(Method.T())
286 resource_list = List.T(Resource.T())
288 def get_children(self):
289 return self.param_list + self.method_list + self.resource_list
292class Application(Element):
293 xmltagname = 'application'
294 guessable_xmlns = [guts_xmlns]
296 doc_list = List.T(Doc.T())
297 grammars = Grammars.T(optional=True)
298 resources_list = List.T(Resources.T())
299 resource_type_list = List.T(ResourceType.T(xmltagname='resource_type'))
300 method_list = List.T(Method.T())
301 representation_list = List.T(Representation.T())
302 param_list = List.T(Param.T())
304 def get_children(self):
305 return self.resources_list + self.resource_type_list \
306 + self.method_list + self.representation_list \
307 + self.param_list
309 def update(self, force=False):
310 if self._hyper is None or force:
311 hyper = dict(
312 resource_type={},
313 resource={},
314 method={},
315 representation={},
316 param={})
318 self._update(hyper)
320 def iter_resources(self):
321 self.update()
322 for rs in self.resources_list:
323 for p, res in rs.iter_resources():
324 yield clean_path(p), res
326 def iter_requests(self):
327 for res_path, res in self.iter_resources():
328 for method in res.iter_methods():
329 if method.request:
330 yield res_path, method.name, method.request
332 def supported_param_names(self, path, method='GET'):
333 path = clean_path(path)
334 for res_path, method_name, request in self.iter_requests():
335 if res_path == path and method_name == method:
336 return [param.name for param in request.param_list]
338 def describe(self, indent=''):
339 lines = []
340 for res_path, res in self.iter_resources():
341 lines.append(indent + res_path)
342 lines.extend(res.describe(indent))
344 return lines
346 def __str__(self):
347 return '\n'.join(self.describe())
350def load_xml(*args, **kwargs):
351 wadl = guts.load_xml(*args, **kwargs)
352 if not isinstance(wadl, Application):
353 FileLoadError('Not a WADL file.')
355 return wadl
358if __name__ == '__main__':
359 import os
360 import sys
361 import urllib.request
363 if os.path.exists(sys.argv[1]):
364 wadl = load_xml(filename=sys.argv[1])
365 else:
366 f = urllib.request.urlopen(sys.argv[1])
367 wadl = load_xml(stream=f)
369 print(wadl)