Method run_sync

tornado/ioloop.py:462–547

Starts the `IOLoop`, runs the given function, and stops the loop. The function must return either an awaitable object or ``None``. If the function returns an awaitable object, the `IOLoop` will run until the awaitable is resolved (and `run_sync()` will return the awa

(self, func: Callable, timeout: Optional[float] = None)

        raise NotImplementedError()

    def run_sync(self, func: Callable, timeout: Optional[float] = None) -> Any:
        """Starts the `IOLoop`, runs the given function, and stops the loop.

        The function must return either an awaitable object or
        ``None``. If the function returns an awaitable object, the
        `IOLoop` will run until the awaitable is resolved (and
        `run_sync()` will return the awaitable's result). If it raises
        an exception, the `IOLoop` will stop and the exception will be
        re-raised to the caller.

        The keyword-only argument ``timeout`` may be used to set
        a maximum duration for the function.  If the timeout expires,
        a `asyncio.TimeoutError` is raised.

        This method is useful to allow asynchronous calls in a
        ``main()`` function::

            async def main():
                # do stuff...

            if __name__ == '__main__':
                IOLoop.current().run_sync(main)

        .. versionchanged:: 4.3
           Returning a non-``None``, non-awaitable value is now an error.

        .. versionchanged:: 5.0
           If a timeout occurs, the ``func`` coroutine will be cancelled.

        .. versionchanged:: 6.2
           ``tornado.util.TimeoutError`` is now an alias to ``asyncio.TimeoutError``.
        """
        if typing.TYPE_CHECKING:
            FutureCell = TypedDict(  # noqa: F841
                "FutureCell", {"future": Optional[Future], "timeout_called": bool}
            )
        future_cell = {"future": None, "timeout_called": False}  # type: FutureCell

        def run() -> None:
            try:
                result = func()
                if result is not None:
                    from tornado.gen import convert_yielded

                    result = convert_yielded(result)
            except Exception:
                fut = Future()  # type: Future[Any]
                future_cell["future"] = fut
                future_set_exc_info(fut, sys.exc_info())
            else:
                if is_future(result):
                    future_cell["future"] = result
                else:
                    fut = Future()
                    future_cell["future"] = fut
                    fut.set_result(result)
            assert future_cell["future"] is not None
            self.add_future(future_cell["future"], lambda future: self.stop())