[embedded-elt] Deprecate Sling factory method and SlingSource and

SlingTarget Resources

python_modules/libraries/dagster-embedded-elt/dagster_embedded_elt/sling/asset_defs.py

@@ -8,12 +8,16 @@
     MaterializeResult,
     multi_asset,
 )
-from dagster._annotations import experimental
+from dagster._annotations import deprecated
+from dagster._utils.warnings import deprecation_warning
 
 from dagster_embedded_elt.sling.resources import SlingMode, SlingResource
 
 
-@experimental
+@deprecated(
+    breaking_version="0.23.0",
+    additional_warn_text="Use `@sling_assets` instead.",
+)
 def build_sling_asset(
     asset_spec: AssetSpec,
     source_stream: str,
@@ -76,6 +80,12 @@ def build_sling_asset(
         required_resource_keys={sling_resource_key},
     )
     def sync(context: AssetExecutionContext) -> MaterializeResult:
+        deprecation_warning(
+            "build_sling_asset",
+            breaking_version="0.23.0",
+            additional_warn_text="Use `@sling_assets` property instead.",
+        )
+
         sling: SlingResource = getattr(context.resources, sling_resource_key)
         last_row_count_observed = None
         for stdout_line in sling.sync(

python_modules/libraries/dagster-embedded-elt/dagster_embedded_elt/sling/resources.py

@@ -18,8 +18,9 @@
     PermissiveConfig,
     get_dagster_logger,
 )
-from dagster._annotations import experimental
+from dagster._annotations import deprecated, experimental
 from dagster._utils.env import environ
+from dagster._utils.warnings import deprecation_warning
 from pydantic import Field
 
 from dagster_embedded_elt.sling.asset_decorator import get_streams_from_replication
@@ -29,6 +30,7 @@
 logger = get_dagster_logger()
 
 ANSI_ESCAPE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
+DEPRECATION_WARNING_TEXT = "{name} has been deprecated, use `SlingConnectionResource` for both source and target connections."
 
 
 class SlingMode(str, Enum):
@@ -44,6 +46,10 @@ class SlingMode(str, Enum):
     BACKFILL = "backfill"
 
 
+@deprecated(
+    breaking_version="0.23.0",
+    additional_warn_text=DEPRECATION_WARNING_TEXT.format(name="SlingSourceConnection"),
+)
 class SlingSourceConnection(PermissiveConfig):
     """A Sling Source Connection defines the source connection used by :py:class:`~dagster_elt.sling.SlingResource`.
 
@@ -77,6 +83,10 @@ class SlingSourceConnection(PermissiveConfig):
     )
 
 
+@deprecated(
+    breaking_version="0.23.0",
+    additional_warn_text=DEPRECATION_WARNING_TEXT.format(name="SlingTargetConnection"),
+)
 class SlingTargetConnection(PermissiveConfig):
     """A Sling Target Connection defines the target connection used by :py:class:`~dagster_elt.sling.SlingResource`.
 
@@ -181,6 +191,22 @@ class SlingResource(ConfigurableResource):
     @contextlib.contextmanager
     def _setup_config(self) -> Generator[None, None, None]:
         """Uses environment variables to set the Sling source and target connections."""
+        if self.source_connection:
+            deprecation_warning(
+                "source_connection",
+                "0.23",
+                "source_connection has been deprecated, provide a list of SlingConnectionResource to the `connections` parameter instead.",
+                stacklevel=4,
+            )
+
+        if self.target_connection:
+            deprecation_warning(
+                "target_connection",
+                "0.23",
+                "target_connection has been deprecated, provide a list of SlingConnectionResource to the `connections` parameter instead.",
+                stacklevel=4,
+            )
+
         sling_source = None
         sling_target = None
         if self.source_connection:
@@ -235,6 +261,10 @@ def _exec_sling_cmd(
             if proc.returncode != 0:
                 raise Exception("Sling command failed with error code %s", proc.returncode)
 
+    @deprecated(
+        breaking_version="0.23.0",
+        additional_warn_text="sync has been deprecated, use `replicate` instead.",
+    )
     def sync(
         self,
         source_stream: str,

python_modules/libraries/dagster-embedded-elt/dagster_embedded_elt_tests/replication_configs/pg_to_duckdb_with_dag_config/sling_dag.py

@@ -1,6 +1,16 @@
-from dagster import Definitions, file_relative_path
-from dagster_embedded_elt.sling import DagsterSlingTranslator, sling_assets
-from dagster_embedded_elt.sling.resources import SlingConnectionResource, SlingResource
+from dagster import AssetSpec, Definitions, file_relative_path
+from dagster_embedded_elt.sling import (
+    DagsterSlingTranslator,
+    SlingMode,
+    build_sling_asset,
+    sling_assets,
+)
+from dagster_embedded_elt.sling.resources import (
+    SlingConnectionResource,
+    SlingResource,
+    SlingSourceConnection,
+    SlingTargetConnection,
+)
 
 replication_config = file_relative_path(__file__, "sling_replication.yaml")
 
@@ -19,6 +29,14 @@
     ]
 )
 
+asset_deprecated = build_sling_asset(
+    asset_spec=AssetSpec(key=["main", "dest_tbl"]),
+    source_stream="file:///tmp/test.csv",
+    target_object="main.dest_table",
+    mode=SlingMode.INCREMENTAL,
+    primary_key="id",
+)
+
 
 @sling_assets(replication_config=replication_config)
 def my_assets(context, sling: SlingResource):
@@ -30,7 +48,17 @@ def my_assets(context, sling: SlingResource):
         context.log.info(row)
 
 
+sling_other_resource = SlingResource(
+    source_connection=SlingSourceConnection(
+        type="postgres",
+        connection_string="postgres://postgres:postgres@localhost:5432/finance?sslmode=disable",
+    ),
+    target_connection=SlingTargetConnection(
+        type="duckdb", connection_string="duckdb:///var/tmp/duckdb.db"
+    ),
+)
+
 defs = Definitions(
-    assets=[my_assets],
-    resources={"sling": sling_resource},
+    assets=[my_assets, asset_deprecated],
+    resources={"sling": sling_resource, "other": sling_other_resource},
 )