Source code for micro.common.discover

import asyncio
import importlib
import logging

from aiolocals import preserve_context_in_threads, job_context
from micro.rest.api import API

from .service import Service
from ._utils import schedule_task_in_thread


log = logging.getLogger(__name__)
__magic = '__api_discover__'


[docs]def api(func):
    """
    Mark the function as API factory

    :param func: Function that returns a micro.rest.api.API inheritor
    """
    setattr(func, __magic, True)
    return func


[docs]def discover_api(module):
    """
    Discover the API initializer in gunicorn

    :param module: A module containing API factory
    :type module: module
    :return: micro.rest.api.API inheritor instance
    """
    for attr in dir(module):
        value = getattr(module, attr)
        if hasattr(value, __magic):
            return value
    else:
        raise RuntimeError("service.py does not declare an API")


def initialize(api_initializer, *args, **kwargs):
    """
    Initialize service using separate thread.
    This is not thread safe and will be run only during service startup. Common use case:


    .. code::

        @api
        @asyncio.coroutine
        def service_factory():
            '''
            :rtype: micro.common.service.Service
            '''
            ...

        if __name__ == '__main__':
            app = initialize(service_factory)
            app.debug(port=8888)
    """
    app = api_initializer(*args, **kwargs)
    if hasattr(api_initializer, 'loop'):
        loop = api_initializer.loop
    else:
        loop = asyncio.get_event_loop()
    @asyncio.coroutine
    def initialize_api(_app):
        if asyncio.iscoroutine(_app):
            _app = yield from _app
        return _app

    initialization_app_task = job_context("GUNICORN API INITIALIZATION")(initialize_api)(app)
    if loop.is_running():
        return initialization_app_task
    else:
        app = loop.run_until_complete(initialization_app_task)
        return app


[docs]def gunicorn(*args, **kwargs):
    """
    Gunicorn entry point.

    :param args: Arguments to pass to your API factory
    :type args: iterable
    :param kwargs: Keyworded arguments to pass to your API factory
    :type kwargs: dict
    :return:  API instance
    :rtype: micro.rest.api.API
    """
    try:
        service = importlib.import_module('service')
    except ImportError:
        print("Service module is unavailable")
        raise SystemExit()
    else:
        api_initializer = discover_api(service)

        app = initialize(api_initializer, *args, **kwargs)

        if isinstance(app, Service):
            return app.production()
        elif isinstance(app, API):
            return app
        else:
            raise RuntimeError("Unable to initialize service: API factory returnable is not API or Service")