"""
This defines a shortcut make_api() for creating self-describing REST APIs
"""
import abc
import os.path
import logging
import asyncio
from aiohttp import hdrs
from micro.web import Response
from micro.web.http import redirect
from .swag import swagger
logger = logging.getLogger(__name__)
def serve_swagger(*args):
return Response(headers={'content-type': 'application/json'}, text=str(swagger))
def serve_swagger_debug_redirect(*args):
redirect("/_debug/index.html?url=/swagger.json")
def serve_health_checker(*args):
return Response(headers={'content-type': 'text/plain'}, text='UP')
class API(metaclass=abc.ABCMeta):
"""
Base class for APIs
"""
@abc.abstractproperty
def name(self):
"""
API name
"""
@abc.abstractproperty
def description(self):
"""
API description
"""
@abc.abstractproperty
def version(self):
"""
API version
"""
def __init__(self, app):
"""
:param app: A Micro.Web application
:type app: micro.web.Micro
"""
self.app = app
swagger.set_api(self)
self.app.router.add_route(
hdrs.METH_GET, "/swagger.json",
serve_swagger,
name='__swagger'
)
self.app.router.add_route(
hdrs.METH_ANY, "/health-check",
serve_health_checker,
name='__health_checker'
)
if self.app.debug:
logging.info("Running API server in debug mode")
self.app.router.add_route(
hdrs.METH_GET, "/_debug", serve_swagger_debug_redirect, name='__debug'
)
swagger_path = os.path.abspath(
os.path.normpath(os.path.join(os.curdir, './_swagger_ui/dist/'))
)
if os.path.exists(swagger_path):
self.app.router.add_static(
'/_debug/', swagger_path
)
else:
logger.warning('Starting in debug mode without Swagger UI') # pragma: no cover
def info(self):
return {
"title": self.name,
"description": self.description,
"version": self.version
}
def make_handler(self, *args, **kwargs):
return self.app.make_handler()
def debug(self, host='0.0.0.0', port=7000, name=None, configure_logging=True):
"""
Run the app in debugging mode. This is only used when the server is started without gunicorn.
This function does not return until the app is shut down.
:param host: A host where to bind. Probably either '0.0.0.0' or 'localhost'.
:param port: A port where to bind
:param name: A specific app name to override the default
:param configure_logging: Pass False if logging formatting should not be setup
"""
if not name:
name = self.name
self.app.global_state.debug = True
self.app.run(host=host, port=port, name=name, configure_logging=configure_logging)
def __call__(self, *args, **kwargs):
"""gunicorn compatibility"""
return self
def finish(self):
return self.app.finish()
[docs]def make_api(app, callbacks=(), name=None, description=None, version=None):
"""
Create API server instance
:param app: micro.core.app.Micro instance
:type app: :class:`micro.web.Micro`
:param callbacks: iterable of callbacks which should be run before API initializes
:type callbacks: iterable
:param name: Name of this API (required)
:type name: str
:param description: Description of this API (required)
:type description: str
:param version: Version of this API (required)
:type version: str
:return: API instance, a :class:`micro.rest.api.API` subclass
"""
# Import here for mockability
from micro.web.utils import maybe_coroutine
required_params = 'name', 'description', 'version'
for param_name in required_params:
if not locals()[param_name]:
raise RuntimeError("Missing requred API parameter %s" % param_name)
api = type(name, (API, ), dict(name=name, description=description, version=version))
for callback in callbacks:
asyncio.async(maybe_coroutine(callback))
return api(app)
__all__ = ['make_api']