✨ Add SQLServer transform flow (#1089)

* ✨ Add SQLServer transform flow

* 📝 Add SQLServer transform prefect flow

---------

Co-authored-by: angelika233 <[email protected]>

docs/references/orchestration/prefect/flows.md

@@ -58,6 +58,8 @@
 
 ::: viadot.orchestration.prefect.flows.sql_server_to_parquet
 
+::: viadot.orchestration.prefect.flows.sql_server_transform
+
 ::: viadot.orchestration.prefect.flows.supermetrics_to_adls
 
 ::: viadot.orchestration.prefect.flows.transform

src/viadot/orchestration/prefect/flows/__init__.py

@@ -30,6 +30,7 @@
 from .sharepoint_to_s3 import sharepoint_to_s3
 from .sql_server_to_minio import sql_server_to_minio
 from .sql_server_to_parquet import sql_server_to_parquet
+from .sql_server_transform import sql_server_transform
 from .supermetrics_to_adls import supermetrics_to_adls
 from .transform import transform
 from .transform_and_catalog import transform_and_catalog
@@ -67,6 +68,7 @@
     "sharepoint_to_s3",
     "sql_server_to_minio",
     "sql_server_to_parquet",
+    "sql_server_transform",
     "supermetrics_to_adls",
     "transform",
     "transform_and_catalog",

src/viadot/orchestration/prefect/flows/sql_server_transform.py

@@ -0,0 +1,34 @@
+"""Flow for transforming data inside the SQLServer."""
+
+from prefect import flow
+
+from viadot.orchestration.prefect.tasks import sql_server_query
+
+
+@flow(
+    name="transform--sql_server",
+    description="Transform data inside the SQLServer.",
+    retries=1,
+    retry_delay_seconds=60,
+)
+def sql_server_transform(
+    query: str,
+    sql_server_credentials_secret: str | None = None,
+    sql_server_config_key: str | None = None,
+) -> None:
+    """Run query inside the SQLServer.
+
+    Args:
+        query (str, required): The query to execute on the SQL Server database.
+            If the qery doesn't start with "SELECT" returns an empty DataFrame.
+        sql_server_credentials_secret (str, optional): The name of the secret storing
+            the credentialsto the SQLServer. Defaults to None.
+            More info on: https://docs.prefect.io/concepts/blocks/
+        sql_server_config_key (str, optional): The key in the viadot config
+            holding relevant credentials to the SQLServer. Defaults to None.
+    """
+    sql_server_query(
+        query=query,
+        config_key=sql_server_config_key,
+        credentials_secret=sql_server_credentials_secret,
+    )