Source code for aiopyramid.config
"""
This module provides view mappers for running views in asyncio.
"""
import asyncio
from pyramid.config.views import DefaultViewMapper
from pyramid.exceptions import ConfigurationError
from .helpers import synchronize, is_generator
[docs]class AsyncioMapperBase(DefaultViewMapper):
"""
Base class for asyncio view mappers.
"""
[docs] def run_in_coroutine_view(self, view):
view = synchronize(view)
def coroutine_view(context, request):
# Trigger loading of post data to avoid synchronization problems
# This must be done in a non-async context
request.params.__getitem__ = request.params.__getitem__
return view(context, request)
return coroutine_view
[docs] def run_in_executor_view(self, view):
synchronizer = synchronize(strict=False)
def executor_view(context, request):
# Trigger loading of post data to avoid synchronization problems
# This must be done in a non-async context
request.params.__getitem__ = request.params.__getitem__
try:
# since we are running in a new thread,
# remove the old wsgi.file_wrapper for uwsgi
request.environ.pop('wsgi.file_wrapper')
finally:
exe = synchronizer(asyncio.get_event_loop().run_in_executor)
return exe(None, view, context, request)
return executor_view
[docs]class CoroutineMapper(AsyncioMapperBase):
def __call__(self, view):
original = view
view = super().__call__(view)
if is_generator(original) or is_generator(
getattr(original, '__call__', None)
):
view = asyncio.coroutine(view)
elif not asyncio.iscoroutinefunction(original):
raise ConfigurationError(
'Non-coroutine {} mapped to coroutine.'.format(original)
)
return self.run_in_coroutine_view(view)
[docs]class ExecutorMapper(AsyncioMapperBase):
def __call__(self, view):
if asyncio.iscoroutinefunction(view) or asyncio.iscoroutinefunction(
getattr(view, '__call__', None)
):
raise ConfigurationError(
'Coroutine {} mapped to executor.'.format(view)
)
view = super().__call__(view)
return self.run_in_executor_view(view)
[docs]class CoroutineOrExecutorMapper(AsyncioMapperBase):
def __call__(self, view):
original = view
while asyncio.iscoroutinefunction(view):
try:
view = view.__wrapped__ # unwrap coroutine
except AttributeError:
break
view = super().__call__(view)
if (
asyncio.iscoroutinefunction(original) or
is_generator(original) or
is_generator(
getattr(original, '__call__', None)
)
):
view = asyncio.coroutine(view)
return self.run_in_coroutine_view(view)
else:
return self.run_in_executor_view(view)