ASGI Support

CHANGELOG.rst

@@ -1,6 +1,13 @@
 CHANGELOG
 =========
 
+6.0.0 (unreleased)
+------------------
+
+- Replaced aiohttp with ASGI (running with uvicorn by default)
+  [dmanchon,masipcat]
+
+
 5.0.0a2 (unreleased)
 --------------------
 

guillotina/api/files.py

@@ -1,6 +1,6 @@
 import mimetypes
 
-from aiohttp.web import StreamResponse
+from guillotina.response import StreamResponse
 from guillotina import configure
 from guillotina._settings import app_settings
 from guillotina.api.content import DefaultOPTIONS
@@ -76,19 +76,18 @@ async def serve_file(self, fi):
         filepath = str(fi.file_path.absolute())
         filename = fi.file_path.name
         with open(filepath, 'rb') as f:
-            resp = StreamResponse()
+            resp = StreamResponse(status=200)
             resp.content_type, _ = mimetypes.guess_type(filename)
 
             disposition = 'filename="{}"'.format(filename)
-            if 'text' not in resp.content_type:
+            if 'text' not in (resp.content_type or ""):
                 disposition = 'attachment; ' + disposition
 
             resp.headers['CONTENT-DISPOSITION'] = disposition
 
             data = f.read()
             resp.content_length = len(data)
             await resp.prepare(self.request)
-
             await resp.write(data)
             await resp.write_eof()
             return resp

guillotina/api/ws.py

@@ -3,7 +3,6 @@
 
 import aiohttp
 import ujson
-from aiohttp import web
 from guillotina import configure
 from guillotina import logger
 from guillotina import routes
@@ -187,7 +186,7 @@ async def handle_ws_request(self, ws, message):
     async def __call__(self):
         tm = get_tm()
         await tm.abort()
-        ws = web.WebSocketResponse()
+        ws = self.request.get_ws()
         await ws.prepare(self.request)
 
         async for msg in ws:

guillotina/asgi.py

@@ -0,0 +1,115 @@
+from guillotina import glogging
+from guillotina import task_vars
+from guillotina.exceptions import ConflictError
+from guillotina.exceptions import TIDConflictError
+from guillotina.request import AsgiStreamReader
+from guillotina.request import Request
+
+
+logger = glogging.getLogger('guillotina')
+
+
+def headers_to_list(headers):
+    return [[k.encode(), v.encode()] for k, v in headers.items()]
+
+
+class AsgiApp:
+    def __init__(self, config_file, settings, loop):
+        self.app = None
+        self.config_file = config_file
+        self.settings = settings
+        self.loop = loop
+        self.on_cleanup = []
+        self.route = None
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] == "http" or scope["type"] == "websocket":
+            return await self.handler(scope, receive, send)
+
+        elif scope["type"] == "lifespan":
+            while True:
+                message = await receive()
+                if message['type'] == 'lifespan.startup':
+                    await self.startup()
+                    await send({'type': 'lifespan.startup.complete'})
+                elif message['type'] == 'lifespan.shutdown':
+                    await self.shutdown()
+                    await send({'type': 'lifespan.shutdown.complete'})
+                    return
+
+    async def startup(self):
+        from guillotina.factory.app import startup_app
+        self.app = await startup_app(
+            config_file=self.config_file,
+            settings=self.settings,
+            loop=self.loop,
+            server_app=self
+        )
+        return self.app
+
+    async def shutdown(self):
+        for clean in self.on_cleanup:
+            await clean(self)
+
+    async def handler(self, scope, receive, send):
+        # Aiohttp compatible StreamReader
+        payload = AsgiStreamReader(receive)
+
+        if scope["type"] == "websocket":
+            scope["method"] = "GET"
+
+        request = Request(
+            scope["scheme"],
+            scope["method"],
+            scope["path"],
+            scope["query_string"],
+            scope["headers"],
+            payload,
+            loop=self.loop,
+            send=send,
+            scope=scope,
+            receive=receive
+        )
+
+        task_vars.request.set(request)
+        resp = await self.request_handler(request)
+
+        # WS handling after view execution missing here!!!
+        if scope["type"] != "websocket":
+            from guillotina.response import StreamResponse
+
+            if not isinstance(resp, StreamResponse):
+                await send(
+                    {
+                        "type": "http.response.start",
+                        "status": resp.status,
+                        "headers": headers_to_list(resp.headers)
+                    }
+                )
+                body = resp.text or ""
+                await send({"type": "http.response.body", "body": body.encode()})
+
+    async def request_handler(self, request, retries=0):
+        try:
+            route = await self.app.router.resolve(request)
+            return await route.handler(request)
+
+        except (ConflictError, TIDConflictError) as e:
+            if self.settings.get('conflict_retry_attempts', 3) > retries:
+                label = 'DB Conflict detected'
+                if isinstance(e, TIDConflictError):
+                    label = 'TID Conflict Error detected'
+                tid = getattr(getattr(request, '_txn', None), '_tid', 'not issued')
+                logger.debug(
+                    f'{label}, retrying request, tid: {tid}, retries: {retries + 1})',
+                    exc_info=True)
+                request._retry_attempt = retries + 1
+                request.clear_futures()
+                return await self.request_handler(request, retries + 1)
+            else:
+                logger.error(
+                    'Exhausted retry attempts for conflict error on tid: {}'.format(
+                        getattr(getattr(request, '_txn', None), '_tid', 'not issued')
+                    ))
+                from guillotina.response import HTTPConflict
+                return HTTPConflict()

guillotina/commands/__init__.py

@@ -178,14 +178,15 @@ def __run_with_monitor(self, app, settings):
             self.__run(app, settings)
 
     def __run(self, app, settings):
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(app.startup())
         if asyncio.iscoroutinefunction(self.run):
             # Blocking call which returns when finished
-            loop = asyncio.get_event_loop()
             loop.run_until_complete(self.run(self.arguments, settings, app))
-            loop.run_until_complete(self.cleanup(app))
-            loop.close()
         else:
-            self.run(self.arguments, settings, app)
+            self.run(self.arguments, settings, app.app)
+        loop.run_until_complete(self.cleanup(app))
+        loop.close()
 
         if self.profiler is not None:
             if self.arguments.profile_output:
@@ -202,8 +203,7 @@ def __run(self, app, settings):
 
     async def cleanup(self, app):
         try:
-            app.freeze()
-            await app.on_cleanup.send(app)
+            await app.shutdown()
         except Exception:
             logger.warning('Unhandled error cleanup tasks', exc_info=True)
         for task in asyncio.Task.all_tasks():
@@ -239,8 +239,7 @@ def signal_handler(self, signal, frame):
     def make_app(self, settings):
         signal.signal(signal.SIGINT, self.signal_handler)
         loop = self.get_loop()
-        return loop.run_until_complete(
-            make_app(settings=settings, loop=loop))
+        return make_app(settings=settings, loop=loop)
 
     def get_parser(self):
         parser = argparse.ArgumentParser(description=self.description)

guillotina/commands/server.py

@@ -1,20 +1,6 @@
-from aiohttp import web
+import uvicorn
 from guillotina.commands import Command
 
-import asyncio
-import sys
-
-try:
-    from aiohttp.web_log import AccessLogger  # type: ignore
-except ImportError:
-    from aiohttp.helpers import AccessLogger  # type: ignore
-
-try:
-    import aiohttp_autoreload
-    HAS_AUTORELOAD = True
-except ImportError:
-    HAS_AUTORELOAD = False
-
 
 class ServerCommand(Command):
     description = 'Guillotina server runner'
@@ -34,21 +20,7 @@ def get_parser(self):
         return parser
 
     def run(self, arguments, settings, app):
-        if arguments.reload:
-            if not HAS_AUTORELOAD:
-                sys.stderr.write(
-                    'You must install aiohttp_autoreload for the --reload option to work.\n'
-                    'Use `pip install aiohttp_autoreload` to install aiohttp_autoreload.\n'
-                )
-                return 1
-            aiohttp_autoreload.start()
-
         port = arguments.port or settings.get('address', settings.get('port'))
         host = arguments.host or settings.get('host', '0.0.0.0')
-        log_format = settings.get('access_log_format', AccessLogger.LOG_FORMAT)
-        try:
-            web.run_app(app, host=host, port=port,
-                        access_log_format=log_format)
-        except asyncio.CancelledError:
-            # server shut down, we're good here.
-            pass
+
+        uvicorn.run(app, host=host, port=port, reload=arguments.reload)

guillotina/factory/__init__.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 from guillotina.factory import serialize  # noqa
 from guillotina.factory.app import make_app  # noqa
+from guillotina.factory.app import startup_app  # noqa
 from guillotina.factory.app import configure_application  # noqa
 from guillotina.factory.app import load_application  # noqa
 from guillotina.factory.content import ApplicationRoot  # noqa

guillotina/factory/app.py

@@ -4,10 +4,8 @@
 import logging.config
 from copy import deepcopy
 
-from aiohttp import web
 from guillotina import configure
 from guillotina import glogging
-from guillotina import task_vars
 from guillotina._settings import app_settings
 from guillotina._settings import default_settings
 from guillotina.behaviors import apply_concrete_behaviors
@@ -23,14 +21,10 @@
 from guillotina.events import ApplicationConfiguredEvent
 from guillotina.events import ApplicationInitializedEvent
 from guillotina.events import DatabaseInitializedEvent
-from guillotina.exceptions import ConflictError
-from guillotina.exceptions import TIDConflictError
 from guillotina.factory.content import ApplicationRoot
 from guillotina.interfaces import IApplication
 from guillotina.interfaces import IDatabase
 from guillotina.interfaces import IDatabaseConfigurationFactory
-from guillotina.request import Request
-from guillotina.response import HTTPConflict
 from guillotina.traversal import TraversalRouter
 from guillotina.utils import lazy_apply
 from guillotina.utils import list_or_dict_items
@@ -119,47 +113,6 @@ def load_application(module, root, settings):
     app_configurator.load_application(module)
 
 
-class GuillotinaAIOHTTPApplication(web.Application):
-    async def _handle(self, request, retries=0):
-        task_vars.request.set(request)
-        try:
-            return await super()._handle(request)
-        except (ConflictError, TIDConflictError) as e:
-            if app_settings.get('conflict_retry_attempts', 3) > retries:
-                label = 'DB Conflict detected'
-                if isinstance(e, TIDConflictError):
-                    label = 'TID Conflict Error detected'
-                tid = getattr(getattr(request, '_txn', None), '_tid', 'not issued')
-                logger.debug(
-                    f'{label}, retrying request, tid: {tid}, retries: {retries + 1})',
-                    exc_info=True)
-                request._retry_attempt = retries + 1
-                request.clear_futures()
-                return await self._handle(request, retries + 1)
-            logger.error(
-                'Exhausted retry attempts for conflict error on tid: {}'.format(
-                    getattr(getattr(request, '_txn', None), '_tid', 'not issued')
-                ))
-            return HTTPConflict()
-
-    def _make_request(self, message, payload, protocol, writer, task,
-                      _cls=Request):
-        return _cls(
-            message, payload, protocol, writer, task,
-            self._loop,
-            client_max_size=self._client_max_size)
-
-
-def make_aiohttp_application():
-    middlewares = [resolve_dotted_name(m) for m in app_settings.get('middlewares', [])]
-    router_klass = app_settings.get('router', TraversalRouter)
-    router = resolve_dotted_name(router_klass)()
-    return GuillotinaAIOHTTPApplication(
-        router=router,
-        middlewares=middlewares,
-        **app_settings.get('aiohttp_settings', {}))
-
-
 _dotted_name_settings = (
     'auth_extractors',
     'auth_token_validators',
@@ -196,7 +149,19 @@ def optimize_settings(settings):
             settings[name] = resolve_dotted_name(new_val)
 
 
-async def make_app(config_file=None, settings=None, loop=None, server_app=None):
+def make_app(config_file=None, settings=None, loop=None):
+    from guillotina.asgi import AsgiApp
+
+    # middlewares = [resolve_dotted_name(m) for m in app_settings.get('middlewares', [])]
+    router_klass = app_settings.get('router', TraversalRouter)
+    router = resolve_dotted_name(router_klass)()
+
+    app = AsgiApp(config_file, settings, loop)
+    app.router = router
+    return app
+
+
+async def startup_app(config_file=None, settings=None, loop=None, server_app=None):
     '''
     Make application from configuration
 
@@ -219,8 +184,8 @@ async def make_app(config_file=None, settings=None, loop=None, server_app=None):
         loop = asyncio.get_event_loop()
 
     if config_file is not None:
-        with open(config_file, 'r') as config:
-            settings = json.load(config)
+        from guillotina.commands import get_settings
+        settings = get_settings(config_file)
     elif settings is None:
         raise Exception('Neither configuration or settings')
 
@@ -277,8 +242,6 @@ async def make_app(config_file=None, settings=None, loop=None, server_app=None):
             app_logger.error('Could not setup logging configuration', exc_info=True)
 
     # Make and initialize aiohttp app
-    if server_app is None:
-        server_app = make_aiohttp_application()
     root.app = server_app
     server_app.root = root
     server_app.config = config
@@ -369,6 +332,7 @@ async def close_utilities(app):
         app_logger.info('Removing ' + key)
         await root.del_async_utility(key)
 
+
 async def close_dbs(app):
     root = get_utility(IApplication, name='root')
     for db in root: