def get_canned_queries(self, database_name, actor):
        queries = self.metadata("queries", database=database_name, fallback=False) or {}
        for more_queries in pm.hook.canned_queries(
            datasette=self, database=database_name, actor=actor,
        ):
            if callable(more_queries):
                more_queries = more_queries()
            if asyncio.iscoroutine(more_queries):
                more_queries = await more_queries
            queries.update(more_queries or {})
        # Fix any {"name": "select ..."} queries to be {"name": {"sql": "select ..."}}
        for key in queries:
            if not isinstance(queries[key], dict):
                queries[key] = {"sql": queries[key]}
            # Also make sure "name" is available:
            queries[key]["name"] = key
        return queries 

def dispatch_consumers(self, event_name: str, agent_id: AgentId,
                                 args: Tuple[Any, ...] = tuple()) -> None:
        log_fmt = 'DISPATCH_CONSUMERS(ev:{}, ag:{})'
        log_args = (event_name, agent_id)
        if self.root_app['config']['debug']['log-events']:
            log.debug(log_fmt, *log_args)
        scheduler = get_scheduler_from_app(self.root_app)
        for consumer in self.consumers[event_name]:
            cb = consumer.callback
            try:
                if asyncio.iscoroutine(cb):
                    await scheduler.spawn(cb)
                elif asyncio.iscoroutinefunction(cb):
                    await scheduler.spawn(cb(consumer.context, agent_id, event_name, *args))
                else:
                    cb = functools.partial(cb, consumer.context, agent_id, event_name, *args)
                    self.loop.call_soon(cb)
            except asyncio.CancelledError:
                raise
            except Exception:
                log.exception(log_fmt + ': unexpected-error', *log_args) 

def dispatch_subscribers(self, event_name: str, agent_id: AgentId,
                                   args: Tuple[Any, ...] = tuple()) -> None:
        log_fmt = 'DISPATCH_SUBSCRIBERS(ev:{}, ag:{})'
        log_args = (event_name, agent_id)
        if self.root_app['config']['debug']['log-events']:
            log.debug(log_fmt, *log_args)
        scheduler = get_scheduler_from_app(self.root_app)
        for subscriber in self.subscribers[event_name]:
            cb = subscriber.callback
            try:
                if asyncio.iscoroutine(cb):
                    await scheduler.spawn(cb)
                elif asyncio.iscoroutinefunction(cb):
                    await scheduler.spawn(cb(subscriber.context, agent_id, event_name, *args))
                else:
                    cb = functools.partial(cb, subscriber.context, agent_id, event_name, *args)
                    self.loop.call_soon(cb)
            except asyncio.CancelledError:
                raise
            except Exception:
                log.exception(log_fmt + ': unexpected-error', *log_args) 

def test_decorator(self, fo, patch_timer):
        """
        time works with asyncio results functions.
        """

        @aio.time(fo)
        async def func():
            await asyncio.sleep(0)
            return 42

        rv = func()

        assert asyncio.iscoroutine(rv)
        assert [] == fo._observed

        rv = await rv

        assert [1] == fo._observed
        assert 42 == rv 

def finish(self):
        callbacks = self._finish_callbacks
        self._finish_callbacks = []

        for (cb, args, kwargs) in callbacks:
            try:
                res = cb(self, *args, **kwargs)
                if (asyncio.iscoroutine(res) or
                        isinstance(res, asyncio.Future)):
                    yield from res
            except Exception as exc:
                self.loop.call_exception_handler({
                    'message': "Error in finish callback",
                    'exception': exc,
                    'application': self,
                }) 

def steamdebug(self, ctx, *, shit: str):
        """This is the part where I make 20,000 typos before I get it right"""
        # "what the fuck is with your variable naming" - EJH2
        # seth seriously what the fuck - Robin
        import asyncio
        import os
        import random
        import re
        from datetime import datetime, timedelta
        try:
            rebug = eval(shit)
            if asyncio.iscoroutine(rebug):
                rebug = await rebug
            await ctx.send(py.format(rebug))
        except Exception as damnit:
            await ctx.send(py.format("{}: {}".format(type(damnit).__name__, damnit))) 

def infodebug(self, ctx, *, shit:str):
        """This is the part where I make 20,000 typos before I get it right"""
        # "what the fuck is with your variable naming" - EJH2
        # seth seriously what the fuck - Robin
        import asyncio
        import os
        import random
        import re
        from datetime import datetime, timedelta
        try:
            rebug = eval(shit)
            if asyncio.iscoroutine(rebug):
                rebug = await rebug
            await ctx.send(py.format(rebug))
        except Exception as damnit:
            await ctx.send(py.format("{}: {}".format(type(damnit).__name__, damnit))) 

def on_packet(self, packet, warn_unknown=True):
        source_address, data = packet
        probable_peer = self.network.get_verified_by_address(source_address)
        if probable_peer:
            probable_peer.last_response = time()
        if self._prefix != data[:22]:
            return
        msg_id = chr(ord(data[22:23]))
        if msg_id in self.decode_map:
            handler = self.decode_map[msg_id]
            try:
                result = handler(source_address, data)
                if iscoroutine(result):
                    self.register_anonymous_task('on_packet', ensure_future(result), ignore=(Exception,))
            except Exception:
                self.logger.error("Exception occurred while handling packet!\n"
                                  + ''.join(format_exception(*sys.exc_info())))
        elif warn_unknown:
            self.logger.warning("Received unknown message: %d from (%s, %d)", ord(msg_id), *source_address) 

def on_linked_e2e(self, source_address, payload, circuit_id):
        if not self.request_cache.has(u"link-request", payload.identifier):
            self.logger.warning("Invalid linked-e2e identifier")
            return

        cache = self.request_cache.pop(u"link-request", payload.identifier)
        circuit = cache.circuit
        circuit.e2e = True
        circuit.hs_session_keys = cache.hs_session_keys
        callback = self.e2e_callbacks.get(cache.info_hash, None)
        if callback:
            result = callback((self.circuit_id_to_ip(circuit.circuit_id), CIRCUIT_ID_PORT))
            if iscoroutine(result):
                self.register_anonymous_task('e2e_callback', result)
        else:
            self.logger.error('On linked e2e: could not find download for %s!', cache.info_hash) 

def consumer(self, fn):
        """Consumer decorator

        :param fn: coroutine consumer function

        Example:

        >>> api = StreamingAPI('my_service_key')
        >>> stream = api.get_stream()

        >>> @stream.consumer
        >>> @asyncio.coroutine
        >>> def handle_event(payload):
        >>>     print(payload)

        """
        if self._consumer_fn is not None:
            raise ValueError('Consumer function is already defined for this '
                             'Stream instance')
        if not any([asyncio.iscoroutine(fn), asyncio.iscoroutinefunction(fn)]):
            raise ValueError('Consumer function must be a coroutine')
        self._consumer_fn = fn 

def verify_jwt(self, a_jwt, audience, leeway=0, **requests_kwargs):
        """Verify if the token is correct

        Returns:
             dict: the claims of the given jwt if verification is successful.

        Raises:
            ValueError: if verification failed.
        """
        key_identifier = key._get_key_id_from_jwt_header(a_jwt)

        public_key = self._retrieve_pub_key(key_identifier, requests_kwargs)
        if asyncio.iscoroutine(public_key):
            public_key = await public_key

        return self._decode_jwt(
            a_jwt, key_identifier, public_key,
            audience=audience, leeway=leeway) 

def test_iscoroutine(self):
        async def foo(): pass

        f = foo()
        try:
            self.assertTrue(asyncio.iscoroutine(f))
        finally:
            f.close() # silence warning

        # Test that asyncio.iscoroutine() uses collections.abc.Coroutine
        class FakeCoro:
            def send(self, value): pass
            def throw(self, typ, val=None, tb=None): pass
            def close(self): pass
            def __await__(self): yield

        self.assertTrue(asyncio.iscoroutine(FakeCoro())) 

def as_future(self, fun, *args, **kwargs):
        try:
            res = fun(*args, **kwargs)
        except Exception:
            return create_future_error(create_failure())
        else:
            if isinstance(res, Future):
                return res
            elif iscoroutine(res):
                return self._loop.create_task(res)
            elif isinstance(res, AsyncGeneratorType):
                raise RuntimeError(
                    "as_future() received an async generator function; does "
                    "'{}' use 'yield' when you meant 'await'?".format(
                        str(fun)
                    )
                )
            else:
                return create_future_success(res) 

def __call__(self, request: Request, response: Response) -> None:
        task: Any
        response_tasks: Union[
            Iterable[Callable[[], None]], Callable[[], None]
        ] = []

        if response.ctx:
            response_tasks = response.ctx.get(
                'background_tasks', response_tasks
            )

        if response_tasks and not isinstance(response_tasks, Iterable):
            response_tasks = (response_tasks,)

        for task in response_tasks:
            if asyncio.iscoroutinefunction(task):
                task = task()

            elif not asyncio.iscoroutine(task):
                future = self.executor.submit(task)
                future.add_done_callback(self.done_callback)
                return

            task = asyncio.create_task(task)
            task.add_done_callback(self.done_callback) 

def __call__(self, title, *command):
        parser = get_argument_parser()
        args, command_args = parser.parse_known_args(command)

        api_method_name = args.api_method_name
        parsed = docopt(args.doc, command_args)
        kwargs = set_kwargs(parsed)
        for k, v in kwargs.items():
            if v and isinstance(v, str) and (v[0], v[-1]) == ('"', '"'):
                kwargs[k] = v[1:-1]
        params = json.dumps({"method": api_method_name, "params": kwargs})

        method = getattr(self.test.daemon, f'jsonrpc_{api_method_name}')
        result = method(**kwargs)
        if asyncio.iscoroutine(result):
            result = await result
        output = jsonrpc_dumps_pretty(result, ledger=self.test.daemon.ledger)
        self.examples.setdefault(api_method_name, []).append({
            'title': title,
            'curl': f"curl -d'{params}' http://localhost:5279/",
            'lbrynet': 'lbrynet ' + ' '.join(command),
            'python': f'requests.post("http://localhost:5279", json={params}).json()',
            'output': output.strip()
        })
        return json.loads(output)['result'] 

def __call__(self, *args, **kwargs):
        import asyncio

        _stories.mounted.instrumented = True
        try:
            c = self.story(*args, **kwargs)
        finally:
            _stories.mounted.instrumented = False
        if not asyncio.iscoroutine(c):
            raise Exception("A coroutine executor expected")  # pragma: no cover
        try:
            c.send(None)
        except StopIteration as error:
            return error.value
        else:
            raise Exception("A coroutine does not return")  # pragma: no cover 

def assert_completion_candidates(self, cmdcls, args, exp_cands, exp_curarg_seps=()):
        cands = cmdcls.completion_candidates(args)
        if asyncio.iscoroutine(cands):
            cands = await cands

        if cands is None:
            self.assertIs(None, exp_cands)
        else:
            if cands and not isinstance(cands[0], str):
                self.assertEqual(tuple(tuple(subcands) for subcands in cands),
                                 tuple(sorted(exp_cands)))
                for subcands,exp_seps in zip(cands,exp_curarg_seps):
                    self.assertEqual(subcands.curarg_seps, exp_seps)
            else:
                self.assertEqual(tuple(cands), tuple(sorted(exp_cands)))
                self.assertEqual(cands.curarg_seps, exp_curarg_seps) 

def __init__(self, future=None):
        def callback(source, target):
            try:
                asyncio.futures._chain_future(source, target)
            except Exception as exc:
                if self.concurrent_future.set_running_or_notify_cancel():
                    self.concurrent_future.set_exception(exc)
                raise

        if asyncio.iscoroutine(future):
            future = asyncio.ensure_future(future, loop=unsync.loop)
        if isinstance(future, concurrent.futures.Future):
            self.concurrent_future = future
            self.future = asyncio.Future(loop=unsync.loop)
            self.future._loop.call_soon_threadsafe(callback, self.concurrent_future, self.future)
        else:
            self.future = future or asyncio.Future(loop=unsync.loop)
            self.concurrent_future = concurrent.futures.Future()
            self.future._loop.call_soon_threadsafe(callback, self.future, self.concurrent_future) 

def _run_test_method(self, method):
        result = method()
        if asyncio.iscoroutine(result):
            self.loop.run_until_complete(
                asyncio.wait_for(result, timeout=self.TEST_TIMEOUT)) 

def invoke_startup(self):
        for hook in pm.hook.startup(datasette=self):
            if callable(hook):
                hook = hook()
            if asyncio.iscoroutine(hook):
                hook = await hook 

def permission_allowed(self, actor, action, resource=None, default=False):
        "Check permissions using the permissions_allowed plugin hook"
        result = None
        for check in pm.hook.permission_allowed(
            datasette=self, actor=actor, action=action, resource=resource,
        ):
            if callable(check):
                check = check()
            if asyncio.iscoroutine(check):
                check = await check
            if check is not None:
                result = check
        used_default = False
        if result is None:
            result = default
            used_default = True
        self._permission_checks.append(
            {
                "when": datetime.datetime.utcnow().isoformat(),
                "actor": actor,
                "action": action,
                "resource": resource,
                "used_default": used_default,
                "result": result,
            }
        )
        return result 

def sync(coro, loop: asyncio.AbstractEventLoop):
        if asyncio.iscoroutine(coro):
            # Run async function in the loop and return the value or raise the exception
            return asyncio.run_coroutine_threadsafe(coro, loop=loop).result()

        return coro 

def execute(coro):
    """
        run a function or coroutine

    Parameters
    ----------
    coro : asyncio.coroutine or function
    """
    if asyncio.iscoroutine(coro):
        return await coro
    else:
        return coro 

def test_can_paginate_iterator(s3_client, bucket_name, create_object):
    for i in range(5):
        key_name = 'key%s' % i
        await create_object(key_name)

    paginator = s3_client.get_paginator('list_objects')
    responses = []
    async for page in paginator.paginate(
            PaginationConfig={'PageSize': 1}, Bucket=bucket_name):
        assert not asyncio.iscoroutine(page)
        responses.append(page)
    assert len(responses) == 5, responses
    data = [r for r in responses]
    key_names = [el['Contents'][0]['Key'] for el in data]
    assert key_names == ['key0', 'key1', 'key2', 'key3', 'key4'] 

def test_still_coroutine_function(self, fo):
        """
        It's ensured that a decorated function still passes as a coroutine
        function.  Otherwise PYTHONASYNCIODEBUG=1 breaks.
        """
        func = aio.time(fo)(coro)
        new_coro = func()

        assert inspect.iscoroutine(new_coro)
        assert inspect.iscoroutinefunction(func)

        await new_coro 

def _on_join_prepare(self, previous_assignment):
        self._subscription.begin_reassignment()
        self._group_subscription = None

        # commit offsets prior to rebalance if auto-commit enabled
        if previous_assignment is not None:
            try:
                await self._maybe_do_last_autocommit(previous_assignment)
            except Errors.KafkaError as err:
                # We would retry any retriable commit already
                log.error("OffsetCommit failed before join, ignoring: %s", err)
            revoked = previous_assignment.tps
        else:
            revoked = set([])

        # execute the user's callback before rebalance
        log.info("Revoking previously assigned partitions %s for group %s",
                 revoked, self.group_id)
        if self._subscription.listener:
            try:
                res = self._subscription.listener.on_partitions_revoked(
                    revoked)
                if asyncio.iscoroutine(res):
                    await res
            except Exception:
                log.exception("User provided subscription listener %s"
                              " for group %s failed on_partitions_revoked",
                              self._subscription.listener, self.group_id) 

def _on_join_complete(
        self, generation, member_id, protocol,
        member_assignment_bytes
    ):
        assignor = self._lookup_assignor(protocol)
        assert assignor, 'invalid assignment protocol: %s' % protocol

        assignment = ConsumerProtocol.ASSIGNMENT.decode(
            member_assignment_bytes)

        # update partition assignment
        self._subscription.assign_from_subscribed(assignment.partitions())

        # give the assignor a chance to update internal state
        # based on the received assignment
        assignor.on_assignment(assignment)

        # We need to start this task before callback to avoid deadlocks.
        # Callback can rely on something like ``Consumer.position()`` that
        # requires committed point to be refreshed.
        await self._stop_commit_offsets_refresh_task()
        self.start_commit_offsets_refresh_task(
            self._subscription.subscription.assignment)

        assigned = set(self._subscription.assigned_partitions())
        log.info("Setting newly assigned partitions %s for group %s",
                 assigned, self.group_id)

        # execute the user's callback after rebalance
        if self._subscription.listener:
            try:
                res = self._subscription.listener.on_partitions_assigned(
                    assigned)
                if asyncio.iscoroutine(res):
                    await res
            except Exception:
                log.exception("User provided listener %s for group %s"
                              " failed on partition assignment: %s",
                              self._subscription.listener, self.group_id,
                              assigned) 

def __init__(
        self,
        coro: Coroutine,
        complete_hook: Callable = None,
        ident: str = None,
        task_future: asyncio.Future = None,
        queued_time: float = None,
    ):
        """
        Initialize the pending task.

        Args:
            coro: The coroutine to be run
            complete_hook: A callback to run on completion
            ident: A string identifier for the task
            task_future: A future to be resolved to the asyncio Task
            queued_time: When the pending task was added to the queue
        """
        if not asyncio.iscoroutine(coro):
            raise ValueError(f"Expected coroutine, got {coro}")
        self._cancelled = False
        self.complete_hook = complete_hook
        self.coro = coro
        self.queued_time: float = queued_time
        self.unqueued_time: float = None
        self.ident = ident or coro_ident(coro)
        self.task_future = task_future or asyncio.get_event_loop().create_future() 

def run(
        self,
        coro: Coroutine,
        task_complete: Callable = None,
        ident: str = None,
        timing: dict = None,
    ) -> asyncio.Task:
        """
        Start executing a coroutine as an async task, bypassing the pending queue.

        Args:
            coro: The coroutine to run
            task_complete: An optional callback to run on completion
            ident: A string identifier for the task
            timing: An optional dictionary of timing information

        Returns: the new asyncio task instance

        """
        if self._cancelled:
            raise RuntimeError("Task queue has been cancelled")
        if not asyncio.iscoroutine(coro):
            raise ValueError(f"Expected coroutine, got {coro}")
        if not ident:
            ident = coro_ident(coro)
        if self.timed:
            if not timing:
                timing = dict()
            coro = coro_timed(coro, timing)
        task = self.loop.create_task(coro)
        return self.add_active(task, task_complete, ident, timing) 

def _notify_and_ensure_future(self, notify):
        tasks = []
        for subscription in self._iterate_subscriptions:
            maybe_coroutine = notify(subscription)
            if asyncio.iscoroutine(maybe_coroutine):
                tasks.append(maybe_coroutine)
        if tasks:
            return asyncio.ensure_future(asyncio.wait(tasks))
        else:
            f = asyncio.get_event_loop().create_future()
            f.set_result(None)
            return f