Add external_access_integration and secret to python model impl

.changes/unreleased/Features-20240402-131330.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add external_access_integration and secret to python models
+time: 2024-04-02T13:13:30.952425+01:00
+custom:
+  Author: Lindblomsebastian
+  Issue: "955"

dbt/adapters/snowflake/impl.py

@@ -205,6 +205,10 @@ def submit_python_job(self, parsed_model: dict, compiled_code: str):
 
         packages = parsed_model["config"].get("packages", [])
         imports = parsed_model["config"].get("imports", [])
+        external_access_integrations = parsed_model["config"].get(
+            "external_access_integrations", []
+        )
+        secrets = parsed_model["config"].get("secrets", {})
         # adding default packages we need to make python model work
         default_packages = ["snowflake-snowpark-python"]
         package_names = [package.split("==")[0] for package in packages]
@@ -213,9 +217,18 @@ def submit_python_job(self, parsed_model: dict, compiled_code: str):
                 packages.append(default_package)
         packages = "', '".join(packages)
         imports = "', '".join(imports)
-        # we can't pass empty imports clause to snowflake
+        external_access_integrations = "', '".join(external_access_integrations)
+        secrets = ", ".join(f"'{key}' = {value}" for key, value in secrets.items())
+
+        # we can't pass empty imports, external_access_integrations or secrets clause to snowflake
         if imports:
             imports = f"IMPORTS = ('{imports}')"
+        if external_access_integrations:
+            # Black is trying to make this a tuple.
+            # fmt: off
+            external_access_integrations = f"EXTERNAL_ACCESS_INTEGRATIONS = ('{external_access_integrations}')"
+        if secrets:
+            secrets = f"SECRETS = ({secrets})"
 
         if self.config.args.SEND_ANONYMOUS_USAGE_STATS:
             snowpark_telemetry_string = "dbtLabs_dbtPython"
@@ -230,6 +243,8 @@ def submit_python_job(self, parsed_model: dict, compiled_code: str):
 LANGUAGE PYTHON
 RUNTIME_VERSION = '{python_version}'
 PACKAGES = ('{packages}')
+{external_access_integrations}
+{secrets}
 {imports}
 HANDLER = 'main'
 EXECUTE AS CALLER

tests/functional/adapter/test_python_model.py

@@ -138,3 +138,69 @@ def teardown_method(self, project):
     def test_custom_target(self, project):
         results = run_dbt()
         assert results[0].node.schema == f"{project.test_schema}_MY_CUSTOM_SCHEMA"
+
+
+EXTERNAL_ACCESS_INTEGRATION_MODE = """
+import pandas
+import snowflake.snowpark as snowpark
+
+def model(dbt, session: snowpark.Session):
+    dbt.config(
+        materialized="table",
+        external_access_integration=["test_external_access_integration"],
+        packages=["httpx==0.26.0"]
+    )
+    import httpx
+    return session.create_dataframe(
+        pandas.DataFrame(
+            [{"result": httpx.get(url="https://www.google.com").stats_code}]
+        )
+    )
+"""
+
+
+class TestExternalAccessIntegration:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"external_access_integration_python_model.py": EXTERNAL_ACCESS_INTEGRATION_MODE}
+
+    def test_external_access_integration(self, project):
+        # The SA running the integration tests does not have the required permissions.
+        # project.run_sql(
+        #     "create or replace network rule test_network_rule type = host_port mode = egress value_list= ('www.google.com:443');"
+        # )
+        # project.run_sql(
+        #     "create or replace external access integration test_external_access_integration allowed_network_rules = (test_network_rule) enabled = true;"
+        # )
+        run_dbt(["run"])
+
+
+SECRETS_MODE = """
+import pandas
+import snowflake.snowpark as snowpark
+
+def model(dbt, session: snowpark.Session):
+    dbt.config(
+        materialized="table",
+        secrets={"secret_value": "test_secret"},
+    )
+    import _snowflake
+    return session.create_dataframe(
+        pandas.DataFrame(
+            [{"secret_value": _snowflake.get_generic_secret_string(test_secret)}]
+        )
+    )
+"""
+
+
+class TestSecrets:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"secret_python_model.py": SECRETS_MODE}
+
+    def test_secrets(self, project):
+        # The SA running the integration tests does not have the required permissions.
+        # project.run_sql(
+        #     "create or replace secret test_secret type = generic_string secret_string='secret value';"
+        # )
+        run_dbt(["run"])