adds finished_at to step info

Author: rudolfix

dlt/common/pipeline.py

@@ -41,6 +41,7 @@
 from dlt.common.schema.typing import TColumnNames, TColumnSchema, TWriteDisposition, TSchemaContract
 from dlt.common.source import get_current_pipe_name
 from dlt.common.storages.load_storage import LoadPackageInfo
+from dlt.common.time import ensure_pendulum_datetime, precise_time
 from dlt.common.typing import DictStrAny, REPattern, StrAny, SupportsHumanize
 from dlt.common.jsonpath import delete_matches, TAnyJsonPath
 from dlt.common.data_writers.writers import DataWriterMetrics, TLoaderFileFormat
@@ -53,24 +54,61 @@ class _StepInfo(NamedTuple):
     """ids of the loaded packages"""
     load_packages: List[LoadPackageInfo]
     """Information on loaded packages"""
-    started_at: datetime.datetime
     first_run: bool
+    started_at: datetime.datetime
+    finished_at: datetime.datetime
+
+
+class StepMetrics(TypedDict):
+    """Metrics for particular package processed in particular pipeline step"""
+
+    started_at: datetime.datetime
+    """Start of package processing"""
+    finished_at: datetime.datetime
+    """End of package processing"""
+
 
+TStepMetricsCo = TypeVar("TStepMetricsCo", bound=StepMetrics, covariant=True)
 
-class StepInfo(SupportsHumanize):
+
+class StepInfo(SupportsHumanize, Generic[TStepMetricsCo]):
     pipeline: "SupportsPipeline"
+    metrics: Dict[str, List[TStepMetricsCo]]
+    """Metrics per load id. If many sources with the same name were extracted, there will be more than 1 element in the list"""
     loads_ids: List[str]
     """ids of the loaded packages"""
     load_packages: List[LoadPackageInfo]
     """Information on loaded packages"""
-    started_at: datetime.datetime
     first_run: bool
 
+    @property
+    def started_at(self) -> datetime.datetime:
+        """Returns the earliest start date of all collected metrics"""
+        if not self.metrics:
+            return None
+        try:
+            return min(m["started_at"] for l_m in self.metrics.values() for m in l_m)
+        except ValueError:
+            return None
+
+    @property
+    def finished_at(self) -> datetime.datetime:
+        """Returns the latest end date of all collected metrics"""
+        if not self.metrics:
+            return None
+        try:
+            return max(m["finished_at"] for l_m in self.metrics.values() for m in l_m)
+        except ValueError:
+            return None
+
     def asdict(self) -> DictStrAny:
         # to be mixed with NamedTuple
         d: DictStrAny = self._asdict()  # type: ignore
         d["pipeline"] = {"pipeline_name": self.pipeline.pipeline_name}
         d["load_packages"] = [package.asdict() for package in self.load_packages]
+        if self.metrics:
+            d["started_at"] = self.started_at
+            d["finished_at"] = self.finished_at
         return d
 
     def __str__(self) -> str:
@@ -91,7 +129,12 @@ def job_metrics_asdict(
 
     def _astuple(self) -> _StepInfo:
         return _StepInfo(
-            self.pipeline, self.loads_ids, self.load_packages, self.started_at, self.first_run
+            self.pipeline,
+            self.loads_ids,
+            self.load_packages,
+            self.first_run,
+            self.started_at,
+            self.finished_at,
         )
 
 
@@ -100,7 +143,7 @@ class ExtractDataInfo(TypedDict):
     data_type: str
 
 
-class ExtractMetrics(TypedDict):
+class ExtractMetrics(StepMetrics):
     schema_name: str
     job_metrics: Dict[str, DataWriterMetrics]
     """Metrics collected per job id during writing of job file"""
@@ -115,19 +158,19 @@ class ExtractMetrics(TypedDict):
 
 
 class _ExtractInfo(NamedTuple):
+    """NamedTuple cannot be part of the derivation chain so we must re-declare all fields to use it as mixin later"""
+
     pipeline: "SupportsPipeline"
     metrics: Dict[str, List[ExtractMetrics]]
-    """Metrics per load id. If many sources with the same name were extracted, there will be more than 1 element in the list"""
     extract_data_info: List[ExtractDataInfo]
     loads_ids: List[str]
     """ids of the loaded packages"""
     load_packages: List[LoadPackageInfo]
     """Information on loaded packages"""
-    started_at: datetime.datetime
     first_run: bool
 
 
-class ExtractInfo(StepInfo, _ExtractInfo):
+class ExtractInfo(StepInfo[ExtractMetrics], _ExtractInfo):  # type: ignore[misc]
     """A tuple holding information on extracted data items. Returned by pipeline `extract` method."""
 
     def asdict(self) -> DictStrAny:
@@ -178,7 +221,10 @@ def asstr(self, verbosity: int = 0) -> str:
         return ""
 
 
-class NormalizeMetrics(TypedDict):
+# reveal_type(ExtractInfo)
+
+
+class NormalizeMetrics(StepMetrics):
     job_metrics: Dict[str, DataWriterMetrics]
     """Metrics collected per job id during writing of job file"""
     table_metrics: Dict[str, DataWriterMetrics]
@@ -192,11 +238,10 @@ class _NormalizeInfo(NamedTuple):
     """ids of the loaded packages"""
     load_packages: List[LoadPackageInfo]
     """Information on loaded packages"""
-    started_at: datetime.datetime
     first_run: bool
 
 
-class NormalizeInfo(StepInfo, _NormalizeInfo):
+class NormalizeInfo(StepInfo[NormalizeMetrics], _NormalizeInfo):  # type: ignore[misc]
     """A tuple holding information on normalized data items. Returned by pipeline `normalize` method."""
 
     @property
@@ -244,8 +289,13 @@ def asstr(self, verbosity: int = 0) -> str:
         return msg
 
 
+class LoadMetrics(StepMetrics):
+    pass
+
+
 class _LoadInfo(NamedTuple):
     pipeline: "SupportsPipeline"
+    metrics: Dict[str, List[LoadMetrics]]
     destination_type: str
     destination_displayable_credentials: str
     destination_name: str
@@ -259,11 +309,10 @@ class _LoadInfo(NamedTuple):
     """ids of the loaded packages"""
     load_packages: List[LoadPackageInfo]
     """Information on loaded packages"""
-    started_at: datetime.datetime
     first_run: bool
 
 
-class LoadInfo(StepInfo, _LoadInfo):
+class LoadInfo(StepInfo[LoadMetrics], _LoadInfo):  # type: ignore[misc]
     """A tuple holding the information on recently loaded packages. Returned by pipeline `run` and `load` methods"""
 
     def asdict(self) -> DictStrAny:
@@ -329,32 +378,38 @@ def __str__(self) -> str:
         return self.asstr(verbosity=1)
 
 
-TStepMetrics = TypeVar("TStepMetrics")
-TStepInfo = TypeVar("TStepInfo", bound=StepInfo)
+TStepMetrics = TypeVar("TStepMetrics", bound=StepMetrics, covariant=False)
+TStepInfo = TypeVar("TStepInfo", bound=StepInfo[StepMetrics])
 
 
 class WithStepInfo(ABC, Generic[TStepMetrics, TStepInfo]):
     """Implemented by classes that generate StepInfo with metrics and package infos"""
 
     _current_load_id: str
     _load_id_metrics: Dict[str, List[TStepMetrics]]
+    _current_load_started: float
     """Completed load ids metrics"""
 
     def __init__(self) -> None:
         self._load_id_metrics = {}
         self._current_load_id = None
+        self._current_load_started = None
 
     def _step_info_start_load_id(self, load_id: str) -> None:
         self._current_load_id = load_id
+        self._current_load_started = precise_time()
         self._load_id_metrics.setdefault(load_id, [])
 
     def _step_info_complete_load_id(self, load_id: str, metrics: TStepMetrics) -> None:
         assert self._current_load_id == load_id, (
             f"Current load id mismatch {self._current_load_id} != {load_id} when completing step"
             " info"
         )
+        metrics["started_at"] = ensure_pendulum_datetime(self._current_load_started)
+        metrics["finished_at"] = ensure_pendulum_datetime(precise_time())
         self._load_id_metrics[load_id].append(metrics)
         self._current_load_id = None
+        self._current_load_started = None
 
     def _step_info_metrics(self, load_id: str) -> List[TStepMetrics]:
         return self._load_id_metrics[load_id]
@@ -368,8 +423,6 @@ def current_load_id(self) -> str:
     def get_step_info(
         self,
         pipeline: "SupportsPipeline",
-        started_at: datetime.datetime = None,
-        completed_at: datetime.datetime = None,
     ) -> TStepInfo:
         """Returns and instance of StepInfo with metrics and package infos"""
         pass

dlt/extract/extract.py

@@ -225,6 +225,8 @@ def _compute_metrics(self, load_id: str, source: DltSource) -> ExtractMetrics:
                 hints[name] = hint
 
         return {
+            "started_at": None,
+            "finished_at": None,
             "schema_name": source.schema.name,
             "job_metrics": {job.job_id(): metrics for job, metrics in job_metrics.items()},
             "table_metrics": table_metrics,
@@ -352,9 +354,7 @@ def commit_packages(self) -> None:
         # all load ids got processed, cleanup empty folder
         self.extract_storage.delete_empty_extract_folder()
 
-    def get_step_info(
-        self, pipeline: SupportsPipeline, started_at: datetime = None, completed_at: datetime = None
-    ) -> ExtractInfo:
+    def get_step_info(self, pipeline: SupportsPipeline) -> ExtractInfo:
         load_ids = list(self._load_id_metrics.keys())
         load_packages: List[LoadPackageInfo] = []
         metrics: Dict[str, List[ExtractMetrics]] = {}
@@ -368,6 +368,5 @@ def get_step_info(
             describe_extract_data(self.original_data),
             load_ids,
             load_packages,
-            started_at,
             pipeline.first_run,
         )

dlt/load/load.py

@@ -8,7 +8,7 @@
 from dlt.common import sleep, logger
 from dlt.common.configuration import with_config, known_sections
 from dlt.common.configuration.accessors import config
-from dlt.common.pipeline import LoadInfo, SupportsPipeline, WithStepInfo
+from dlt.common.pipeline import LoadInfo, LoadMetrics, SupportsPipeline, WithStepInfo
 from dlt.common.schema.utils import get_child_tables, get_top_level_table
 from dlt.common.storages.load_storage import LoadPackageInfo, ParsedLoadJobFileName, TJobState
 from dlt.common.runners import TRunMetrics, Runnable, workermethod, NullExecutor
@@ -47,7 +47,7 @@
 )
 
 
-class Load(Runnable[Executor], WithStepInfo[str, LoadInfo]):
+class Load(Runnable[Executor], WithStepInfo[LoadMetrics, LoadInfo]):
     pool: Executor
 
     @with_config(spec=LoaderConfiguration, sections=(known_sections.LOAD,))
@@ -347,7 +347,7 @@ def complete_package(self, load_id: str, schema: Schema, aborted: bool = False)
         self.load_storage.complete_load_package(load_id, aborted)
         # TODO: Load must provide a clear interface to get last loads and metrics
         # TODO: get more info ie. was package aborted, schema name etc.
-        self._step_info_complete_load_id(load_id, metrics=None)
+        self._step_info_complete_load_id(load_id, metrics={"started_at": None, "finished_at": None})
         logger.info(
             f"All jobs completed, archiving package {load_id} with aborted set to {aborted}"
         )
@@ -563,12 +563,11 @@ def run(self, pool: Optional[Executor]) -> TRunMetrics:
     def get_step_info(
         self,
         pipeline: SupportsPipeline,
-        started_at: datetime.datetime = None,
-        completed_at: datetime.datetime = None,
     ) -> LoadInfo:
         # TODO: LoadInfo should hold many datasets
         load_ids = list(self._load_id_metrics.keys())
         load_packages: List[LoadPackageInfo] = []
+        metrics: Dict[str, List[LoadMetrics]] = {}
         # get load packages and dataset_name from the last package
         _dataset_name: str = None
         for load_id in self._load_id_metrics.keys():
@@ -579,9 +578,11 @@ def get_step_info(
                     load_package.schema
                 )
             load_packages.append(load_package)
+            metrics[load_id] = self._step_info_metrics(load_id)
 
         return LoadInfo(
             pipeline,
+            metrics,
             Destination.normalize_type(self.initial_client_config.destination_type),
             str(self.initial_client_config),
             self.initial_client_config.destination_name,
@@ -601,6 +602,5 @@ def get_step_info(
             _dataset_name,
             list(load_ids),
             load_packages,
-            started_at,
             pipeline.first_run,
         )

dlt/normalize/normalize.py

@@ -316,6 +316,8 @@ def spool_files(
         self._step_info_complete_load_id(
             load_id,
             {
+                "started_at": None,
+                "finished_at": None,
                 "job_metrics": {job.job_id(): metrics for job, metrics in job_metrics.items()},
                 "table_metrics": {
                     table_name: sum(map(lambda pair: pair[1], metrics), EMPTY_DATA_WRITER_METRICS)
@@ -388,8 +390,6 @@ def get_load_package_info(self, load_id: str) -> LoadPackageInfo:
     def get_step_info(
         self,
         pipeline: SupportsPipeline,
-        started_at: datetime.datetime = None,
-        completed_at: datetime.datetime = None,
     ) -> NormalizeInfo:
         load_ids = list(self._load_id_metrics.keys())
         load_packages: List[LoadPackageInfo] = []
@@ -398,6 +398,4 @@ def get_step_info(
             load_package = self.get_load_package_info(load_id)
             load_packages.append(load_package)
             metrics[load_id] = self._step_info_metrics(load_id)
-        return NormalizeInfo(
-            pipeline, metrics, load_ids, load_packages, started_at, pipeline.first_run
-        )
+        return NormalizeInfo(pipeline, metrics, load_ids, load_packages, pipeline.first_run)

dlt/pipeline/exceptions.py

@@ -1,6 +1,6 @@
 from typing import Any, Dict
 from dlt.common.exceptions import PipelineException
-from dlt.common.pipeline import StepInfo, SupportsPipeline
+from dlt.common.pipeline import StepInfo, StepMetrics, SupportsPipeline
 from dlt.pipeline.typing import TPipelineStep
 
 
@@ -56,7 +56,7 @@ def __init__(
         step: TPipelineStep,
         load_id: str,
         exception: BaseException,
-        step_info: StepInfo = None,
+        step_info: StepInfo[StepMetrics] = None,
     ) -> None:
         self.pipeline = pipeline
         self.step = step