refactor: allow passing of custom loggers into pipeline objects

Author: georgeRobertson

src/dve/pipeline/duckdb_pipeline.py

@@ -1,5 +1,6 @@
 """DuckDB implementation for `Pipeline` object."""
 
+import logging
 from typing import Optional
 
 from duckdb import DuckDBPyConnection, DuckDBPyRelation
@@ -30,6 +31,7 @@ def __init__(
         submitted_files_path: Optional[URI],
         reference_data_loader: Optional[type[BaseRefDataLoader]] = None,
         job_run_id: Optional[int] = None,
+        logger: Optional[logging.Logger] = None,
     ):
         self._connection = connection
         super().__init__(
@@ -41,6 +43,7 @@ def __init__(
             submitted_files_path,
             reference_data_loader,
             job_run_id,
+            logger,
         )
 
     # pylint: disable=arguments-differ

src/dve/pipeline/pipeline.py

@@ -1,6 +1,7 @@
 # pylint: disable=protected-access,too-many-instance-attributes,too-many-arguments,line-too-long
 """Generic Pipeline object to define how DVE should be interacted with."""
 import json
+import logging
 import re
 from collections import defaultdict
 from collections.abc import Generator, Iterable, Iterator
@@ -57,6 +58,7 @@ def __init__(
         submitted_files_path: Optional[URI],
         reference_data_loader: Optional[type[BaseRefDataLoader]] = None,
         job_run_id: Optional[int] = None,
+        logger: Optional[logging.Logger] = None,
     ):
         self._submitted_files_path = submitted_files_path
         self._processed_files_path = processed_files_path
@@ -66,11 +68,16 @@ def __init__(
         self._audit_tables = audit_tables
         self._data_contract = data_contract
         self._step_implementations = step_implementations
-        self._logger = get_logger(__name__)
+        self._logger = logger or get_logger(__name__)
         self._summary_lock = Lock()
         self._rec_tracking_lock = Lock()
         self._aggregates_lock = Lock()
 
+        if self._data_contract:
+            self._data_contract.logger = self._logger
+        if self._step_implementations:
+            self._step_implementations.logger = self._logger
+
     @property
     def job_run_id(self) -> Optional[int]:
         """Unique Identifier for the job/process that is running this Pipeline."""

src/dve/pipeline/spark_pipeline.py

@@ -1,5 +1,6 @@
 """Spark implementation for `Pipeline` object."""
 
+import logging
 from concurrent.futures import Executor
 from typing import Optional
 
@@ -32,6 +33,7 @@ def __init__(
         reference_data_loader: Optional[type[BaseRefDataLoader]] = None,
         spark: Optional[SparkSession] = None,
         job_run_id: Optional[int] = None,
+        logger: Optional[logging.Logger] = None,
     ):
         self._spark = spark if spark else SparkSession.builder.getOrCreate()
         super().__init__(
@@ -43,6 +45,7 @@ def __init__(
             submitted_files_path,
             reference_data_loader,
             job_run_id,
+            logger,
         )
 
     # pylint: disable=arguments-differ