/usr/lib/python2.7/dist-packages/wsme/protocol.py is in python-wsme 0.6-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | import weakref
import pkg_resources
__all__ = [
'CallContext',
'register_protocol', 'getprotocol',
]
registered_protocols = {}
def _cfg(f):
cfg = getattr(f, '_cfg', None)
if cfg is None:
f._cfg = cfg = {}
return cfg
class expose(object):
def __init__(self, path, content_type):
self.path = path
self.content_type = content_type
def __call__(self, func):
func.exposed = True
cfg = _cfg(func)
cfg['content-type'] = self.content_type
cfg.setdefault('paths', []).append(self.path)
return func
class CallContext(object):
def __init__(self, request):
self._request = weakref.ref(request)
self.path = None
self.func = None
self.funcdef = None
@property
def request(self):
return self._request()
class ObjectDict(object):
def __init__(self, obj):
self.obj = obj
def __getitem__(self, name):
return getattr(self.obj, name)
class Protocol(object):
name = None
displayname = None
content_types = []
def resolve_path(self, path):
if '$' in path:
from string import Template
s = Template(path)
path = s.substitute(ObjectDict(self))
return path
def iter_routes(self):
for attrname in dir(self):
attr = getattr(self, attrname)
if getattr(attr, 'exposed', False):
for path in _cfg(attr)['paths']:
yield self.resolve_path(path), attr
def accept(self, request):
return request.headers.get('Content-Type') in self.content_types
def iter_calls(self, request):
pass
def extract_path(self, context):
pass
def read_arguments(self, context):
pass
def encode_result(self, context, result):
pass
def encode_sample_value(self, datatype, value, format=False):
return ('none', 'N/A')
def encode_sample_params(self, params, format=False):
return ('none', 'N/A')
def encode_sample_result(self, datatype, value, format=False):
return ('none', 'N/A')
def register_protocol(protocol):
registered_protocols[protocol.name] = protocol
def getprotocol(name, **options):
protocol_class = registered_protocols.get(name)
if protocol_class is None:
for entry_point in pkg_resources.iter_entry_points(
'wsme.protocols', name):
if entry_point.name == name:
protocol_class = entry_point.load()
if protocol_class is None:
raise ValueError("Cannot find protocol '%s'" % name)
registered_protocols[name] = protocol_class
return protocol_class(**options)
|