Add a dummy ThreadPool implementation for unavailable `multiprocess…

…ing.pool`

pronto/parsers/base.py

@@ -1,6 +1,5 @@
 import abc
 import functools
-import multiprocessing.pool
 import operator
 import os
 import typing
@@ -9,6 +8,7 @@
 
 from ..logic.lineage import Lineage
 from ..ontology import Ontology
+from ..utils.pool import ThreadPool
 
 
 class BaseParser(abc.ABC):
@@ -27,6 +27,10 @@ def parse_from(
  ) -> None:
  return NotImplemented
 
+ @classmethod
+ def pool(cls, threads: int) -> ThreadPool:
+ return ThreadPool(threads)
+
  @classmethod
  def process_import(
  cls,
@@ -66,7 +70,7 @@ def process_imports(
  basepath=basepath,
  timeout=timeout,
  )
- with multiprocessing.pool.ThreadPool(threads) as pool:
+ with cls.pool(threads) as pool:
  return dict(pool.map(lambda i: (i, process(i)), imports))
 
  _entities = {

pronto/parsers/obo.py

@@ -1,4 +1,3 @@
-import multiprocessing.pool
 import os
 
 import fastobo
@@ -38,7 +37,7 @@ def parse_from(self, handle, threads=None):
  # Extract frames from the current document.
  with typechecked.disabled():
  try:
- with multiprocessing.pool.ThreadPool(threads) as pool:
+ with self.pool(threads) as pool:
  pool.map(self.extract_entity, doc)
  except SyntaxError as s:
  location = self.ont.path, s.lineno, s.offset, s.text

pronto/parsers/obojson.py

@@ -1,4 +1,3 @@
-import multiprocessing.pool
 import os
 
 import fastobo
@@ -39,7 +38,7 @@ def parse_from(self, handle, threads=None):
  # Extract frames from the current document.
  with typechecked.disabled():
  try:
- with multiprocessing.pool.ThreadPool(threads) as pool:
+ with self.pool(threads) as pool:
  pool.map(self.extract_entity, doc)
  except SyntaxError as err:
  location = self.ont.path, err.lineno, err.offset, err.text

pronto/utils/pool.py

@@ -0,0 +1,35 @@
+import typing
+from typing import Callable, Iterable, List
+
+try:
+ from multiprocessing.pool import ThreadPool as _ThreadPool
+except ImportError:
+ _ThreadPool = None # type: ignore
+
+
+_T = typing.TypeVar("_T")
+_U = typing.TypeVar("_U")
+
+
+class ThreadPool(object):
+
+ def __init__(self, threads: int = 0):
+ self.threads = threads
+ self.pool = None if _ThreadPool is None else _ThreadPool(self.threads)
+
+ def __enter__(self) -> "Pool":
+ if self.pool is not None:
+ self.pool.__enter__()
+ return self
+
+ def __exit__(self, exc_val, exc_ty, tb):
+ if self.pool is not None:
+ return self.pool.__exit__(exc_val, exc_ty, tb)
+ return False
+
+ def map(self, func: Callable[[_T], _U], items: Iterable[_T]) -> List[_U]:
+ if self.pool is None:
+ return list(map(func, items))
+ else:
+ return self.pool.map(func, items)
+