Source code for aiohttp_apiset.swagger.operations
import importlib
from collections.abc import Mapping
from importlib import import_module
import yaml
from ..utils import import_obj
[docs]class OperationIdMapping(Mapping):
def __init__(self, *args, **kwargs):
super().__init__()
self._operations = []
self.add(*args, **kwargs)
def __getitem__(self, key):
for om in self._operations:
try:
if isinstance(om, Mapping):
return om[key]
else:
return getattr(om, key)
except (KeyError, AttributeError):
pass
raise KeyError(key)
def __iter__(self):
raise NotImplementedError()
def __len__(self):
return len(self._operations)
@classmethod
def _from_str(cls, name):
try:
return import_module(name)
except ImportError:
pass
try:
return import_obj(name)
except AttributeError:
pass
raise ImportError(name)
[docs] def add(self, *args, **kwargs):
""" Add new mapping from args and kwargs
>>> om = OperationIdMapping()
>>> om.add(
... OperationIdMapping(),
... 'aiohttp_apiset.swagger.operations', # any module
... getPets='mymod.handler',
... getPet='mymod.get_pet',
... )
>>> om['getPets']
'mymod.handler'
:param args: str, Mapping, module or obj
:param kwargs: operationId='handler' or operationId=handler
"""
for arg in args:
if isinstance(arg, str):
self._operations.append(self._from_str(arg))
else:
self._operations.append(arg)
if kwargs:
self._operations.append(kwargs)
def get_docstring_swagger(handler):
if isinstance(handler, str):
h = handler
p = []
while isinstance(h, str):
try:
h = importlib.import_module(h)
break
except ImportError:
if '.' not in h:
raise ImportError(handler)
h, t = h.rsplit('.', 1)
p.append(t)
continue
for i in reversed(p):
h = getattr(h, i, None)
if h is None:
raise ImportError(handler)
docstr = h.__doc__
else:
docstr = handler.__doc__
if docstr:
ds = docstr.rsplit(' ---', maxsplit=1)
if len(ds) == 1:
return
swagger_yaml = ds[-1]
operation = yaml.load(swagger_yaml)
if isinstance(operation, dict):
return operation