code: def _execute_and_fetch(self, session: SessionType = None, **kw):
      from ...session import ExecutionInfo, SyncSession, fetch
  
      session = _get_session(self, session)
      fetch_kwargs = kw.pop("fetch_kwargs", dict())
      if session in self._executed_sessions:
          # if has been executed, fetch directly.
          return self.fetch(session=session, **fetch_kwargs)
      ret = self.execute(session=session, **kw)
      if isinstance(ret, ExecutionInfo):
          # wait=False
          aio_task = ret.aio_task
  
          async def _wait():
              await aio_task
  
          def run():
              asyncio.run_coroutine_threadsafe(
                  _wait(), loop=ret.loop
              ).result()
              return fetch(self, session=session, **fetch_kwargs)
  
          return SyncSession._execution_pool.submit(run)
      else:
          # wait=True
          return self.fetch(session=session, **fetch_kwargs)
  
imports: import asyncio
  import atexit
  import concurrent.futures
  import queue
  import threading
  from typing import List
  from weakref import WeakKeyDictionary, ref
  from ...lib.aio import get_isolation
  from ...typing import SessionType, TileableType
  from ..mode import enter_mode
label_desc: Use image node_id 1 for calling the _ExecuteAndFetchMixin obj's underlying member method code with example usage: obj._execute_and_fetch(session) and returns: self, SyncSession, self, fetch