Always fallback on NullExecutor

dlt/load/load.py

@@ -29,6 +29,7 @@
 
 
 class Load(Runnable[Executor]):
+    pool: Executor
 
     @with_config(spec=LoaderConfiguration, sections=(known_sections.LOAD,))
     def __init__(
@@ -48,7 +49,7 @@ def __init__(
         self.destination = destination
         self.capabilities = destination.capabilities()
         self.staging_destination = staging_destination
-        self.pool: Executor = NullExecutor()
+        self.pool = NullExecutor()
         self.load_storage: LoadStorage = self.create_storage(is_storage_owner)
         self._processed_load_ids: Dict[str, str] = {}
         """Load ids to dataset name"""
@@ -380,9 +381,9 @@ def load_single_package(self, load_id: str, schema: Schema) -> None:
                 self.complete_package(load_id, schema, True)
                 raise
 
-    def run(self, pool: Executor) -> TRunMetrics:
+    def run(self, pool: Optional[Executor]) -> TRunMetrics:
         # store pool
-        self.pool = pool
+        self.pool = pool or NullExecutor()
 
         logger.info("Running file loading")
         # get list of loads and order by name ASC to execute schema updates

dlt/normalize/normalize.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Callable, List, Dict, Sequence, Tuple, Set
+from typing import Any, Callable, List, Dict, Sequence, Tuple, Set, Optional
 from concurrent.futures import Future, ProcessPoolExecutor, Executor
 # from multiprocessing.pool import AsyncResult, Pool as ProcessPool
 
@@ -35,12 +35,13 @@
 
 
 class Normalize(Runnable[Executor]):
+    pool: Executor
     @with_config(spec=NormalizeConfiguration, sections=(known_sections.NORMALIZE,))
     def __init__(self, collector: Collector = NULL_COLLECTOR, schema_storage: SchemaStorage = None, config: NormalizeConfiguration = config.value) -> None:
         self.config = config
         self.collector = collector
         self.normalize_storage: NormalizeStorage = None
-        self.pool: Executor = NullExecutor()
+        self.pool = NullExecutor()
         self.load_storage: LoadStorage = None
         self.schema_storage: SchemaStorage = None
         self._row_counts: TRowCount = {}
@@ -279,9 +280,9 @@ def spool_schema_files(self, load_id: str, schema_name: str, files: Sequence[str
 
         return load_id
 
-    def run(self, pool: Executor) -> TRunMetrics:
+    def run(self, pool: Optional[Executor]) -> TRunMetrics:
         # keep the pool in class instance
-        self.pool = pool
+        self.pool = pool or NullExecutor()
         self._row_counts = {}
         logger.info("Running file normalizing")
         # list files and group by schema name, list must be sorted for group by to actually work