def async_endpoint(name):
def add(f):
log.debug('Adding async runtime endpoint {} ({})'.format(f.__name__, name))
if not inspect.iscoroutinefunction(f):
log.error('{} is not a coroutine'.format(f.__name__))
elif name in async_endpoints:
log.error('{} already exists in table'.format(name))
else:
async_endpoints[name] = f
return f
return add
python类iscoroutinefunction()的实例源码
def __new__(cls, name, bases, members):
for key, value in dict(members).items():
if key.startswith('test_') and inspect.iscoroutinefunction(value):
members['async_' + key] = value
members[key] = _make_asyncwrapper(value)
return type.__new__(cls, name, bases, members)
def iscoroutinefunction(func):
"""Return True if func is a decorated coroutine function."""
return (getattr(func, '_is_coroutine', None) is _is_coroutine or
_inspect_iscoroutinefunction(func))
def __init__(cls, name, bases, dct):
super(MethodCollection, cls).__init__(name, bases, dct)
# Build the list of remote methods.
remote_methods = {}
for name, obj in dct.items():
if is_remote_method(obj):
if not inspect.iscoroutinefunction(obj):
raise RuntimeError("Remote method {} is not a coroutine"
.format(obj))
remote_methods[name] = obj
cls.REMOTE_METHODS = remote_methods
def iscoroutinefunction(func):
"""Return True if func is a decorated coroutine function."""
return (getattr(func, '_is_coroutine', False) or
_inspect_iscoroutinefunction(func))
def setup_custom_handlers(custom_handlers):
"""Set up default and custom handlers for JSON API application."""
from . import handlers as default_handlers
from .common import logger
handlers = {
name: handler
for name, handler in inspect.getmembers(default_handlers,
inspect.iscoroutinefunction)
if name in default_handlers.__all__
}
if custom_handlers is not None:
if isinstance(custom_handlers, MutableMapping):
custom_handlers_iter = custom_handlers.items()
elif isinstance(custom_handlers, Sequence):
custom_handlers_iter = ((c.__name__, c) for c in custom_handlers)
else:
raise TypeError('Wrong type of "custom_handlers" parameter. '
'Mapping or Sequence is expected.')
for name, custom_handler in custom_handlers_iter:
handler_name = custom_handler.__name__
if name not in handlers:
logger.warning('Custom handler %s is ignored.', name)
continue
if not inspect.iscoroutinefunction(custom_handler):
logger.error('"%s" is not a co-routine function (ignored).',
handler_name)
continue
handlers[name] = custom_handler
logger.debug('Default handler "%s" is replaced '
'with co-routine "%s" (%s)',
name, handler_name, inspect.getmodule(custom_handler))
return handlers
def on(selector: str, event: str, filter_selector: str=None, throttle=False, stop_propagation=False):
"""Register a coroutine function as a handler for the given event.
Corresponding to `$(selector).on(event, filter_selector, ...)` in jQuery
Args:
selector: A jQuery selector expression
event: One or more space-separated event types and optional namespaces, such as "click".
filter_selector: a selector string to filter the descendants of the selected elements that trigger the event.
If this argument is omitted, the event is always triggered when it reaches the selected element.
throttle: Whether to discard events when Portkey is busy.
Should set to true if the event is going to fire at very high frequency.
If set to false, all events will be queued and handled.
stop_propagation: Prevents the event from bubbling up the DOM tree,
preventing any parent handlers from being notified of the event.
Returns: A decorator that returns the original function.
"""
def decorator(func):
assert inspect.iscoroutinefunction(func), \
'Only a coroutine function (a function defined with an async def syntax)' \
' can be registered as an event handler.'
handler_info = get_handler_info(func)
# assert handler_info.handler is None, \
# 'You should only register one event on a handler function.'
handler_info._handler = func
handler_info.handler = id(func)
handler_info.selector = selector
handler_info.event = event
handler_info.filter_selector = filter_selector
handler_info.throttle = throttle
handler_info.stop_propagation = stop_propagation
return func
return decorator
def preprocess_jquery_arguments(x):
if isinstance(x, html_tag):
return str(x)
if inspect.iscoroutinefunction(x):
on(None, None)(x)
return id2handlerInfo[id(x)]
return x
def addCleanup(self, function, *args, **kwargs):
'''
Add a function, with arguments, to be called when the test is
completed. If function is a coroutine function, it will be run by the
event loop before it is destroyed.
'''
if asyncio.iscoroutinefunction(function):
return super().addCleanup(self.loop.run_until_complete,
function(*args, **kwargs))
return super().addCleanup(function, *args, **kwargs)
def _setUp(self):
''' Create a new loop for each test case '''
asyncio.set_event_loop_policy(self.loop_policy)
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
if asyncio.iscoroutinefunction(self.setUp):
self.loop.run_until_complete(self.setUp())
else:
self.setUp()
def __init__(cls, name, bases, methods):
coros = {}
for base in reversed(cls.__mro__):
coros.update((name, val) for name, val in vars(base).items()
if inspect.iscoroutinefunction(val))
for name, val in vars(cls).items():
if name in coros and not inspect.iscoroutinefunction(val):
raise TypeError('Must use async def %s%s' % (name, inspect.signature(val)))
super().__init__(name, bases, methods)
def __new__(meta, clsname, bases, attributes):
if '__init__' in attributes and not inspect.iscoroutinefunction(attributes['__init__']):
raise TypeError('__init__ must be a coroutine')
return super().__new__(meta, clsname, bases, attributes)
def __call__(self, function):
if not self.shouldCheck:
return function
hints = get_type_hints(function)
def precheck(*args, **kwargs):
all_args = kwargs.copy()
all_args.update(dict(zip(function.__code__.co_varnames, args)))
for argument, argument_type in ((i, type(j)) for i, j in all_args.items()):
if argument in hints:
if not issubclass(argument_type, hints[argument]):
raise TypeError('Type of {} is {} and not {}'.
format(argument,
argument_type,
hints[argument]))
def postcheck(result):
if 'return' in hints:
if not isinstance(result, hints['return']):
raise TypeError('Type of result is {} and not {}'.
format(type(result), hints['return']))
return result
if inspect.iscoroutinefunction(function):
async def type_checker(*args, **kwargs):
precheck(*args, **kwargs)
result = await function(*args, **kwargs)
return postcheck(result)
else:
def type_checker(*args, **kwargs):
precheck(*args, **kwargs)
result = function(*args, **kwargs)
return postcheck(result)
return type_checker
def pytest_pycollect_makeitem(collector, name, obj):
if collector.funcnamefilter(name) and inspect.iscoroutinefunction(obj):
item = pytest.Function(name, parent=collector)
if 'curio' in item.keywords:
return list(collector._genfunctions(name, obj))
def _create_callback_registration(key):
def _inner(self, func):
if not inspect.iscoroutinefunction(func):
raise ValueError("You can only register coroutines!")
self._message_callbacks.setdefault(key, []).append(func)
return func
return _inner
def on_connect(self, func):
if not inspect.iscoroutinefunction(func):
raise ValueError("You can only register coroutines!")
self._connection_callbacks.append(func)
def on_disconnect(self, func):
# Registers a coroutine to be ran right before exit.
# This is so you can modify your state to be JSON-compliant.
if inspect.iscoroutinefunction(func):
raise ValueError("You can only register non-coroutines!")
self._disconnection_callbacks.append(func)
def on_command(self, command, arg_amount=ANY_ARGUMENTS):
"""
Creates a decorator that registers a command handler.
The argument command must include the prefix.
The command handler takes as arguments:
1. The bot instance
2. The command sender.
3. The command recipient, usually a channel.
4. Any arguments that came with the command,
split depending on the arg_amount argument.
As an example, to register a command that looks like this:
!slap nickname
You'd write something like this:
@bot.on_command("!slap", arg_amount=1)
def slap(self, sender, recipient, slappee):
...
"""
def _inner(func):
if not inspect.iscoroutinefunction(func):
raise ValueError("You can only register coroutines!")
self._command_callbacks.setdefault(command, [])\
.append((func, arg_amount))
return func
return _inner
def iscoroutinefunction(f):
return False
# getargspec has been deprecated in Python 3.5
def iscoroutinefunction(func):
# type: (Callable) -> bool
return False
