|
import re
from pyrocko import guts
from pyrocko.guts import make_typed_list_class, String, StringChoice, List, \
Int, Object, Union, Bool, Defer
from pyrocko.io.io_common import FileLoadError
ResourceTypeList = make_typed_list_class(String)
guts_prefix = 'wadl'
guts_xmlns = 'http://wadl.dev.java.net/2009/02'
re_rmsite = re.compile(r'https?://[^/]+')
re_multisl = re.compile(r'/+')
def clean_path(p):
p = re_rmsite.sub('', p)
p = re_multisl.sub('/', p)
if not p.startswith('/'):
p = '/' + p
return p
class HTTPMethods(StringChoice):
choices = [
'GET',
'POST',
'PUT',
'HEAD',
'DELETE']
UriList = make_typed_list_class(String)
StatusCodeList = make_typed_list_class(Int)
class ParamStyle(StringChoice):
choices = [
'plain',
'query',
'matrix',
'header',
'template']
class Doc(Object):
xmltagname = 'doc'
title = String.T(optional=True, xmlstyle='attribute')
class Method2(Union):
members = [HTTPMethods, String]
class Include(Object):
xmltagname = 'include'
href = String.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
class Option(Object):
xmltagname = 'option'
value = String.T(xmlstyle='attribute')
media_type = String.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
class Link(Object):
xmltagname = 'link'
resource_type = String.T(
optional=True, xmlstyle='attribute', xmltagname='resource_type')
rel = String.T(optional=True, xmlstyle='attribute')
rev = String.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
class Grammars(Object):
xmltagname = 'grammars'
doc_list = List.T(Doc.T())
include_list = List.T(Include.T())
class DerefError(Exception):
pass
class Element(Object):
def __init__(self, *args, **kwargs):
Object.__init__(self, *args, **kwargs)
self._hyper = None
def _update(self, hyper):
self._hyper = hyper
id_ = getattr(self, 'id', None)
if id_:
hyper[self.xmltagname][id_] = self
for child in self.get_children():
child._update(hyper)
def get_children(self):
raise NotImplementedError()
def deref(self):
if not self._hyper:
raise Exception('Must call _update() before calling deref()')
obj = self
seen = set()
while True:
seen.add(obj)
href = getattr(obj, 'href', None)
if href:
try:
obj = obj._hyper[obj.xmltagname][href.lstrip('#')]
if obj in seen:
raise DerefError('cyclic reference')
except KeyError:
raise DerefError(href)
else:
return obj
class Param(Element):
xmltagname = 'param'
href = String.T(optional=True, xmlstyle='attribute')
name = String.T(optional=True, xmlstyle='attribute')
style = ParamStyle.T(optional=True, xmlstyle='attribute')
id = String.T(optional=True, xmlstyle='attribute')
type = String.T(default='xs:string', optional=True, xmlstyle='attribute')
default = String.T(optional=True, xmlstyle='attribute')
required = Bool.T(default='false', optional=True, xmlstyle='attribute')
repeating = Bool.T(default='false', optional=True, xmlstyle='attribute')
fixed = String.T(optional=True, xmlstyle='attribute')
path = String.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
option_list = List.T(Option.T())
link = Link.T(optional=True)
def describe(self, indent):
return indent + self.name
def get_children(self):
return []
class Representation(Element):
xmltagname = 'representation'
id = String.T(optional=True, xmlstyle='attribute')
element = String.T(optional=True, xmlstyle='attribute')
media_type = String.T(optional=True, xmlstyle='attribute')
href = String.T(optional=True, xmlstyle='attribute')
profile = UriList.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
param_list = List.T(Param.T())
def get_children(self):
return self.param_list
class Request(Element):
xmltagname = 'request'
doc_list = List.T(Doc.T())
param_list = List.T(Param.T())
representation_list = List.T(Representation.T())
def iter_params(self):
for param in self.param_list:
param = param.deref()
yield param
def describe(self, indent):
lines = []
for param in self.iter_params():
lines.append(param.describe(indent))
return lines
def get_children(self):
return self.param_list + self.representation_list
class Response(Element):
xmltagname = 'response'
status = StatusCodeList.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
param_list = List.T(Param.T())
representation_list = List.T(Representation.T())
def get_children(self):
return self.param_list + self.representation_list
class Method(Element):
xmltagname = 'method'
id = String.T(optional=True, xmlstyle='attribute')
name = String.T(optional=True, xmlstyle='attribute')
href = String.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
request = Request.T(optional=True)
response_list = List.T(Response.T())
def describe(self, indent):
lines = [indent + self.name]
if self.request:
lines.extend(self.request.describe(' ' + indent))
return lines
def get_children(self):
return ([self.request] if self.request else []) + self.response_list
class Resource(Element):
xmltagname = 'resource'
id = String.T(optional=True, xmlstyle='attribute')
type = String.T(optional=True, xmlstyle='attribute')
query_type = String.T(
default='application/x-www-form-urlencoded',
optional=True,
xmlstyle='attribute')
path = String.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
param_list = List.T(Param.T())
method_list = List.T(Method.T())
resource_list = List.T(Defer('Resource.T'))
def iter_resources(self):
yield self.path, self
for res in self.resource_list:
yield self.path + '/' + res.path, res
def iter_methods(self):
for method in self.method_list:
method = method.deref()
yield method
def describe(self, indent):
lines = []
for met in self.iter_methods():
lines.extend(met.describe(' ' + indent))
return lines
def get_children(self):
return self.param_list + self.method_list + self.resource_list
class Resources(Element):
xmltagname = 'resources'
base = String.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
resource_list = List.T(Resource.T())
def iter_resources(self):
for res in self.resource_list:
for p, sr in res.iter_resources():
yield self.base + '/' + p, sr
def get_children(self):
return self.resource_list
class ResourceType(Element):
xmltagname = 'resource_type'
id = String.T(optional=True, xmlstyle='attribute')
doc_list = List.T(Doc.T())
param_list = List.T(Param.T())
method_list = List.T(Method.T())
resource_list = List.T(Resource.T())
def get_children(self):
return self.param_list + self.method_list + self.resource_list
class Application(Element):
xmltagname = 'application'
guessable_xmlns = [guts_xmlns]
doc_list = List.T(Doc.T())
grammars = Grammars.T(optional=True)
resources_list = List.T(Resources.T())
resource_type_list = List.T(ResourceType.T(xmltagname='resource_type'))
method_list = List.T(Method.T())
representation_list = List.T(Representation.T())
param_list = List.T(Param.T())
def get_children(self):
return self.resources_list + self.resource_type_list \
+ self.method_list + self.representation_list \
+ self.param_list
def update(self, force=False):
if self._hyper is None or force:
hyper = dict(
resource_type={},
resource={},
method={},
representation={},
param={})
self._update(hyper)
def iter_resources(self):
self.update()
for rs in self.resources_list:
for p, res in rs.iter_resources():
yield clean_path(p), res
def iter_requests(self):
for res_path, res in self.iter_resources():
for method in res.iter_methods():
if method.request:
yield res_path, method.name, method.request
def supported_param_names(self, path, method='GET'):
path = clean_path(path)
for res_path, method_name, request in self.iter_requests():
if res_path == path and method_name == method:
return [param.name for param in request.param_list]
def describe(self, indent=''):
lines = []
for res_path, res in self.iter_resources():
lines.append(indent + res_path)
lines.extend(res.describe(indent))
return lines
def __str__(self):
return '\n'.join(self.describe())
def load_xml(*args, **kwargs):
wadl = guts.load_xml(*args, **kwargs)
if not isinstance(wadl, Application):
FileLoadError('Not a WADL file.')
return wadl
if __name__ == '__main__':
import os
import sys
import urllib.request
if os.path.exists(sys.argv[1]):
wadl = load_xml(filename=sys.argv[1])
else:
f = urllib.request.urlopen(sys.argv[1])
wadl = load_xml(stream=f)
print(wadl)
|