[python] Ingest somacore classes

.pre-commit-config.yaml

@@ -18,7 +18,8 @@ repos:
         # Pandas 2.x types (e.g. `pd.Series[Any]`). See `_types.py` or https://github.com/single-cell-data/TileDB-SOMA/issues/2839
         # for more info.
         - "pandas-stubs>=2"
-        - "somacore==1.0.23"
+        # Temporary, for PR: see https://github.com/single-cell-data/SOMA/pull/244
+        - "git+https://github.com/single-cell-data/soma@rw/abcs"
         - types-setuptools
       args: ["--config-file=apis/python/pyproject.toml", "apis/python/src", "apis/python/devtools"]
       pass_filenames: false

apis/python/setup.py

@@ -336,8 +336,8 @@ def run(self):
         "pyarrow",
         "scanpy>=1.9.2",
         "scipy",
-        # Note: the somacore version is in .pre-commit-config.yaml too
-        "somacore==1.0.23",
+        # Temporary, for PR: see https://github.com/single-cell-data/SOMA/pull/244
+        "somacore @ git+https://github.com/single-cell-data/soma@rw/abcs",
         "typing-extensions",  # Note "-" even though `import typing_extensions`
     ],
     extras_require={

apis/python/src/tiledbsoma/_eager_iter.py

@@ -0,0 +1,51 @@
+from concurrent import futures
+from typing import Iterator, Optional, TypeVar
+
+_T = TypeVar("_T")
+
+
+class EagerIterator(Iterator[_T]):
+    def __init__(
+        self,
+        iterator: Iterator[_T],
+        pool: Optional[futures.Executor] = None,
+    ):
+        super().__init__()
+        self.iterator = iterator
+        self._pool = pool or futures.ThreadPoolExecutor()
+        self._own_pool = pool is None
+        self._preload_future = self._pool.submit(self.iterator.__next__)
+
+    def __next__(self) -> _T:
+        stopped = False
+        try:
+            if self._preload_future.cancel():
+                # If `.cancel` returns True, cancellation was successful.
+                # The self.iterator.__next__ call has not yet been started,
+                # and will never be started, so we can compute next ourselves.
+                # This prevents deadlocks if the thread pool is too small
+                # and we can never create a preload thread.
+                return next(self.iterator)
+            # `.cancel` returned false, so the preload is already running.
+            # Just wait for it.
+            return self._preload_future.result()
+        except StopIteration:
+            self._cleanup()
+            stopped = True
+            raise
+        finally:
+            if not stopped:
+                # If we have more to do, go for the next thing.
+                self._preload_future = self._pool.submit(self.iterator.__next__)
+
+    def _cleanup(self) -> None:
+        if self._own_pool:
+            self._pool.shutdown()
+
+    def __del__(self) -> None:
+        # Ensure the threadpool is cleaned up in the case where the
+        # iterator is not exhausted. For more information on __del__:
+        # https://docs.python.org/3/reference/datamodel.html#object.__del__
+        self._cleanup()
+        super_del = getattr(super(), "__del__", lambda: None)
+        super_del()

apis/python/src/tiledbsoma/_experiment.py

@@ -9,13 +9,13 @@
 from typing import Optional
 
 from somacore import experiment, query
-from typing_extensions import Self
 
 from . import _tdb_handles
 from ._collection import Collection, CollectionBase
 from ._dataframe import DataFrame
 from ._indexer import IntIndexer
 from ._measurement import Measurement
+from ._query import ExperimentAxisQuery
 from ._scene import Scene
 from ._soma_object import AnySOMAObject
 
@@ -83,13 +83,11 @@ def axis_query(  # type: ignore
         *,
         obs_query: Optional[query.AxisQuery] = None,
         var_query: Optional[query.AxisQuery] = None,
-    ) -> query.ExperimentAxisQuery[Self]:  # type: ignore
+    ) -> ExperimentAxisQuery:
         """Creates an axis query over this experiment.
         Lifecycle: Maturing.
         """
-        # mypy doesn't quite understand descriptors so it issues a spurious
-        # error here.
-        return query.ExperimentAxisQuery(  # type: ignore
+        return ExperimentAxisQuery(
             self,
             measurement_name,
             obs_query=obs_query or query.AxisQuery(),