From 481a7cb5e147c9cca5d5ffc6d36b65520ab0e516 Mon Sep 17 00:00:00 2001
From: Steinthor Palsson <steini90@gmail.com>
Date: Wed, 15 Nov 2023 19:58:05 -0500
Subject: [PATCH] Add deprecation warning for credentials argument

---
 dlt/__init__.py              |  2 ++
 dlt/pipeline/__init__.py     |  3 +++
 dlt/pipeline/deprecations.py | 20 ++++++++++++++++++++
 dlt/pipeline/pipeline.py     |  6 ++++++
 4 files changed, 31 insertions(+)
 create mode 100644 dlt/pipeline/deprecations.py

diff --git a/dlt/__init__.py b/dlt/__init__.py
index f5dde3f204..728343bdd6 100644
--- a/dlt/__init__.py
+++ b/dlt/__init__.py
@@ -31,6 +31,7 @@
 from dlt.extract.decorators import source, resource, transformer, defer
 from dlt.pipeline import pipeline as _pipeline, run, attach, Pipeline, dbt, current as _current, mark as _mark
 from dlt.pipeline import progress
+from dlt import destinations
 
 pipeline = _pipeline
 current = _current
@@ -64,4 +65,5 @@
     "TSecretValue",
     "TCredentials",
     "sources",
+    "destinations",
 ]
diff --git a/dlt/pipeline/__init__.py b/dlt/pipeline/__init__.py
index 3faad8f5a0..af7dd12294 100644
--- a/dlt/pipeline/__init__.py
+++ b/dlt/pipeline/__init__.py
@@ -13,6 +13,7 @@
 from dlt.pipeline.configuration import PipelineConfiguration, ensure_correct_pipeline_kwargs
 from dlt.pipeline.pipeline import Pipeline
 from dlt.pipeline.progress import _from_name as collector_from_name, TCollectorArg, _NULL_COLLECTOR
+from dlt.pipeline.deprecations import credentials_argument_deprecated
 
 
 @overload
@@ -104,6 +105,8 @@ def pipeline(
     # is any of the arguments different from defaults
     has_arguments = bool(orig_args[0]) or any(orig_args[1].values())
 
+    credentials_argument_deprecated("pipeline", credentials, destination)
+
     if not has_arguments:
         context = Container()[PipelineContext]
         # if pipeline instance is already active then return it, otherwise create a new one
diff --git a/dlt/pipeline/deprecations.py b/dlt/pipeline/deprecations.py
new file mode 100644
index 0000000000..138167c8d3
--- /dev/null
+++ b/dlt/pipeline/deprecations.py
@@ -0,0 +1,20 @@
+import typing as t
+import warnings
+
+from dlt.common.destination import Destination, TDestinationReferenceArg
+
+
+def credentials_argument_deprecated(
+    caller_name: str, credentials: t.Optional[t.Any], destination: TDestinationReferenceArg = None
+) -> None:
+    if credentials is None:
+        return
+
+    dest_name = Destination.to_name(destination) if destination else "postgres"
+
+    warnings.warn(
+        f"The `credentials argument` to {caller_name} is deprecated and will be removed in a future version. "
+        f"Pass the same credentials to the `destination` instance instead, e.g. {caller_name}(destination=dlt.destinations.{dest_name}(credentials=...))",
+        DeprecationWarning,
+        stacklevel=2,
+    )
diff --git a/dlt/pipeline/pipeline.py b/dlt/pipeline/pipeline.py
index cf90cf1ff2..1061953569 100644
--- a/dlt/pipeline/pipeline.py
+++ b/dlt/pipeline/pipeline.py
@@ -52,6 +52,7 @@
 from dlt.pipeline.state_sync import STATE_ENGINE_VERSION, load_state_from_destination, merge_state_if_changed, migrate_state, state_resource, json_encode_state, json_decode_state
 
 from dlt.common.schema.utils import normalize_schema_name
+from dlt.pipeline.deprecations import credentials_argument_deprecated
 
 
 def with_state_sync(may_extract_state: bool = False) -> Callable[[TFun], TFun]:
@@ -342,6 +343,9 @@ def load(
         # set destination and default dataset if provided
         self._set_destinations(destination, None)
         self._set_dataset_name(dataset_name)
+
+        credentials_argument_deprecated("pipeline.load", credentials, destination)
+
         self.credentials = credentials or self.credentials
 
         # check if any schema is present, if not then no data was extracted
@@ -449,6 +453,8 @@ def run(
         self._set_destinations(destination, staging)
         self._set_dataset_name(dataset_name)
 
+        credentials_argument_deprecated("pipeline.run", credentials, self.destination)
+
         # sync state with destination
         if self.config.restore_from_destination and not self.full_refresh and not self._state_restored and (self.destination or destination):
             self.sync_destination(destination, staging, dataset_name)