Changed EventStore's async_put and async_get to use thread pool

executor to do mapping between stored events and domain events.

Because it's mostly CPU-bound work that runs in C-extensions. See
note in Python docs about CPU-bound functions that release the GIL:
https://docs.python.org/3/library/asyncio-task.html#running-in-threads

eventsourcing/application.py

@@ -64,8 +64,8 @@ def get(self, aggregate_id: UUID, version: Optional[int] = None) -> TAggregate:
                 lte=version,
             )
             try:
-                snapshot = next(snapshots)
-            except StopIteration:
+                snapshot = snapshots[0]
+            except IndexError:
                 pass
             else:
                 gt = snapshot.originator_version
@@ -110,8 +110,8 @@ async def async_get(
                 lte=version,
             )
             try:
-                snapshot = next(snapshots)
-            except StopIteration:
+                snapshot = snapshots[0]
+            except IndexError:
                 pass
             else:
                 gt = snapshot.originator_version

eventsourcing/persistence.py

@@ -2,18 +2,20 @@
 import os
 import uuid
 from abc import ABC, abstractmethod
+from asyncio import get_running_loop
 from dataclasses import dataclass
 from datetime import datetime
 from decimal import Decimal
 from typing import (
     Any,
+    Callable,
     Dict,
     Generic,
-    Iterator,
     List,
     Mapping,
     Optional,
     Type,
+    TypeVar,
     cast,
 )
 from uuid import UUID
@@ -590,12 +592,7 @@ def put(self, events: List[TDomainEvent], **kwargs: Any) -> None:
         Stores domain events in aggregate sequence.
         """
         self.recorder.insert_events(
-            list(
-                map(
-                    self.mapper.from_domain_event,
-                    events,
-                )
-            ),
+            self._from_domain_events(events),
             **kwargs,
         )
 
@@ -604,12 +601,7 @@ async def async_put(self, events: List[TDomainEvent], **kwargs: Any) -> None:
         Stores domain events in aggregate sequence using asyncio.
         """
         await self.recorder.async_insert_events(
-            list(
-                map(
-                    self.mapper.from_domain_event,
-                    events,
-                )
-            ),
+            await _to_thread(self._from_domain_events, events),
             **kwargs,
         )
 
@@ -620,19 +612,14 @@ def get(
         lte: Optional[int] = None,
         desc: bool = False,
         limit: Optional[int] = None,
-    ) -> Iterator[TDomainEvent]:
+    ) -> List[TDomainEvent]:
         """
         Retrieves domain events from aggregate sequence.
         """
-        return map(
-            self.mapper.to_domain_event,
+        return self._to_domain_events(
             self.recorder.select_events(
-                originator_id=originator_id,
-                gt=gt,
-                lte=lte,
-                desc=desc,
-                limit=limit,
-            ),
+                originator_id=originator_id, gt=gt, lte=lte, desc=desc, limit=limit
+            )
         )
 
     async def async_get(
@@ -642,12 +629,12 @@ async def async_get(
         lte: Optional[int] = None,
         desc: bool = False,
         limit: Optional[int] = None,
-    ) -> Iterator[TDomainEvent]:
+    ) -> List[TDomainEvent]:
         """
         Retrieves domain events from aggregate sequence.
         """
-        return map(
-            self.mapper.to_domain_event,
+        return await _to_thread(
+            self._to_domain_events,
             await self.recorder.async_select_events(
                 originator_id=originator_id,
                 gt=gt,
@@ -657,6 +644,24 @@ async def async_get(
             ),
         )
 
+    def _from_domain_events(self, events: List[TDomainEvent]) -> List[StoredEvent]:
+        return list(map(self.mapper.from_domain_event, events))
+
+    def _to_domain_events(self, stored_events: List[StoredEvent]) -> List[TDomainEvent]:
+        return list(
+            map(
+                self.mapper.to_domain_event,
+                stored_events,
+            )
+        )
+
+
+_T = TypeVar("_T")
+
+
+async def _to_thread(func: Callable[..., _T], *args: Any) -> _T:
+    return await get_running_loop().run_in_executor(None, func, *args)
+
 
 class InfrastructureFactory(ABC):
     """

eventsourcing/system.py

@@ -34,7 +34,7 @@ class ProcessEvent:
 
     def __init__(self, tracking: Optional[Tracking] = None):
         """
-        Initalises the process event with the given tracking object.
+        Initialises the process event with the given tracking object.
         """
         self.tracking = tracking
         self.events: List[AggregateEvent] = []

eventsourcing/tests/test_snapshotting.py

@@ -56,7 +56,7 @@ def test(self):
 
         # Get snapshot.
         snapshots = snapshot_store.get(account.id, desc=True, limit=1)
-        snapshot = next(snapshots)
+        snapshot = snapshots[0]
         assert isinstance(snapshot, Snapshot)
 
         # Reconstruct the bank account.