async_trio_to_trio.py

class TrioPortal:

    def __init__(self, trio_token=None):
        if trio_token is None:
            trio_token = _core.current_trio_token()
        self._trio_token = trio_token

    # This is the part that runs in the trio thread
    def _run_cb_async(self, afn, args, task, token):
        @_core.disable_ki_protection
        async def unprotected_afn():
            return await afn(*args)

        async def await_in_trio_thread_task():
            result = await _core.Result.acapture(unprotected_afn)
            try:
                token.run_sync_soon(_core.reschedule, task, result)
            except trio.RunFinishedError:
                # The entire run finished, so our particular tasks are certainly
                # long gone - it must have cancelled.
                pass

        _core.spawn_system_task(await_in_trio_thread_task, name=afn)

    def _run_sync_cb_async(self, fn, args, task, token):
        @_core.disable_ki_protection
        def unprotected_fn():
            return fn(*args)

        result = _core.Result.capture(unprotected_fn)
        try:
            token.run_sync_soon(_core.reschedule, task, result)
        except trio.RunFinishedError:
            # The entire run finished, so our particular tasks are certainly
            # long gone - it must have cancelled.
            pass

    @_core.enable_ki_protection
    async def _do_it_async(self, cb, fn, args):
        await _core.checkpoint_if_cancelled()
        self._trio_token.run_sync_soon(
            (cb, fn, args, _core.current_task(), _core.current_trio_token()))

        def abort(raise_cancel):
            return _core.Abort.FAILED
        return await _core.wait_task_rescheduled(abort)

    async def run(self, afn, *args):
        return await self._do_it_async(self._run_cb_async, afn, args)

    async def run_sync(self, fn, *args):
        return await self._do_it_async(self._run_sync_cb_async, fn, args)