Method run_in_greenlet

lib/sqlalchemy/util/concurrency.py:305–319

Run sync function in greenlet. Support nested calls

(
        self, fn: Callable[..., _T], *args: Any, **kwargs: Any
    )

        return self.runner.run(fn(*args, **kwargs))

    def run_in_greenlet(
        self, fn: Callable[..., _T], *args: Any, **kwargs: Any
    ) -> _T:
        """Run sync function in greenlet. Support nested calls"""
        _concurrency_shim._initialize(raise_=False)

        if _concurrency_shim._has_greenlet:
            if self.runner.get_loop().is_running():
                # allow for a wrapped test function to call another
                assert in_greenlet()
                return fn(*args, **kwargs)
            else:
                return self.runner.run(greenlet_spawn(fn, *args, **kwargs))
        else:
            return fn(*args, **kwargs)

    def close(self) -> None:
        self.runner.close()