スレッド系のモジュールの無い環境でも使えるようにする

Author: gottadiveintopython

src/asyncgui_ext/clock.py

@@ -8,8 +8,6 @@
 from functools import partial
 from dataclasses import dataclass
 from contextlib import AbstractAsyncContextManager
-from threading import Thread
-from concurrent.futures import ThreadPoolExecutor
 
 from asyncgui import Cancelled, Task, move_on_when, _sleep_forever, _current_task
 
@@ -393,28 +391,9 @@ async def run_in_thread(self, func, *, daemon=None, polling_interval) -> Awaitab
 
             return_value = await clock.run_in_thread(func, polling_interval=...)
         '''
-        return_value = None
-        exception = None
-        done = False
-
-        def wrapper():
-            nonlocal return_value, done, exception
-            try:
-                return_value = func()
-            except Exception as e:
-                exception = e
-            finally:
-                done = True
+        raise NotImplementedError(r"'Clock.run_in_thread()' is not available because 'threading.Thread' is unavailable.")
 
-        Thread(target=wrapper, daemon=daemon, name="asyncgui_ext.clock.Clock.run_in_thread").start()
-        async with _repeat_sleeping(self, polling_interval) as sleep:
-            while not done:
-                await sleep()
-        if exception is not None:
-            raise exception
-        return return_value
-
-    async def run_in_executor(self, executer: ThreadPoolExecutor, func, *, polling_interval) -> Awaitable:
+    async def run_in_executor(self, executer, func, *, polling_interval) -> Awaitable:
         '''
         Runs a function within a :class:`concurrent.futures.ThreadPoolExecutor`, and waits for the completion of the
         function.
@@ -424,30 +403,7 @@ async def run_in_executor(self, executer: ThreadPoolExecutor, func, *, polling_i
             executor = ThreadPoolExecutor()
             return_value = await clock.run_in_executor(executor, func, polling_interval=...)
         '''
-        return_value = None
-        exception = None
-        done = False
-
-        def wrapper():
-            nonlocal return_value, done, exception
-            try:
-                return_value = func()
-            except Exception as e:
-                exception = e
-            finally:
-                done = True
-
-        future = executer.submit(wrapper)
-        try:
-            async with _repeat_sleeping(self, polling_interval) as sleep:
-                while not done:
-                    await sleep()
-        except Cancelled:
-            future.cancel()
-            raise
-        if exception is not None:
-            raise exception
-        return return_value
+        raise NotImplementedError(r"'Clock.run_executor()' is not available because 'concurrent.futures.ThreadPoolExecutor' is unavailable.")
 
     def _update(setattr, zip, min, obj, duration, transition, output_seq_type, anim_params, task, p_time, dt):
         time = p_time[0] + dt
@@ -553,3 +509,69 @@ def __aenter__(self, _current_task=_current_task) -> Awaitable[Callable[[], Awai
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         self._event.cancel()
+
+
+try:
+    from threading import Thread
+except ImportError:
+    pass
+else:
+    async def run_in_thread(clock: Clock, func, *, daemon=None, polling_interval) -> Awaitable:
+        return_value = None
+        exception = None
+        done = False
+
+        def wrapper():
+            nonlocal return_value, done, exception
+            try:
+                return_value = func()
+            except Exception as e:
+                exception = e
+            finally:
+                done = True
+
+        Thread(target=wrapper, daemon=daemon, name="asyncgui_ext.clock.Clock.run_in_thread").start()
+        async with _repeat_sleeping(clock, polling_interval) as sleep:
+            while not done:
+                await sleep()
+        if exception is not None:
+            raise exception
+        return return_value
+
+    run_in_thread.__doc__ = Clock.run_in_thread.__doc__
+    Clock.run_in_thread = run_in_thread
+
+
+try:
+    from concurrent.futures import ThreadPoolExecutor
+except ImportError:
+    pass
+else:
+    async def run_in_executor(clock: Clock, executer: ThreadPoolExecutor, func, *, polling_interval) -> Awaitable:
+        return_value = None
+        exception = None
+        done = False
+
+        def wrapper():
+            nonlocal return_value, done, exception
+            try:
+                return_value = func()
+            except Exception as e:
+                exception = e
+            finally:
+                done = True
+
+        future = executer.submit(wrapper)
+        try:
+            async with _repeat_sleeping(clock, polling_interval) as sleep:
+                while not done:
+                    await sleep()
+        except Cancelled:
+            future.cancel()
+            raise
+        if exception is not None:
+            raise exception
+        return return_value
+
+    run_in_executor.__doc__ = Clock.run_in_executor.__doc__
+    Clock.run_in_executor = run_in_executor