Make conn_id parameter match the DbtDocsS3Operator

cosmos/operators/local.py

@@ -572,13 +572,13 @@ class DbtFreshnessS3LocalOperator(DbtFreshnessLocalOperator):
 
     def __init__(
         self,
-        aws_conn_id: str,
+        connection_id: str,
         bucket_name: str,
         folder_dir: str | None = None,
         **kwargs: str,
     ) -> None:
         "Initializes the operator."
-        self.aws_conn_id = aws_conn_id
+        self.connection_id = connection_id
         self.bucket_name = bucket_name
         self.folder_dir = folder_dir
 
@@ -591,15 +591,15 @@ def upload_to_s3(self, project_dir: str) -> None:
         "Uploads the generated documentation to S3."
         logger.info(
             'Attempting to upload generated docs to S3 using S3Hook("%s")',
-            self.aws_conn_id,
+            self.connection_id,
         )
 
         from airflow.providers.amazon.aws.hooks.s3 import S3Hook
 
         target_dir = f"{project_dir}/target"
 
         hook = S3Hook(
-            self.aws_conn_id,
+            self.connection_id,
             extra_args={
                 "ContentType": "text/html",
             },

dev/dags/dbt_docs.py

@@ -20,6 +20,7 @@
 from cosmos.operators import (
     DbtDocsAzureStorageOperator,
     DbtDocsS3Operator,
+    DbtFreshnessS3Operator,
     DbtDocsGCSOperator,
 )
 from cosmos.profiles import PostgresUserPasswordProfileMapping
@@ -83,6 +84,14 @@ def which_upload():
         bucket_name="cosmos-docs",
     )
 
+    generate_dbt_freshness_aws = DbtFreshnessS3Operator(
+        task_id="generate_dbt_freshness_aws",
+        project_dir=DBT_ROOT_PATH / "jaffle_shop",
+        profile_config=profile_config,
+        connection_id=S3_CONN_ID,
+        bucket_name="cosmos-docs",
+    )
+
     generate_dbt_docs_azure = DbtDocsAzureStorageOperator(
         task_id="generate_dbt_docs_azure",
         project_dir=DBT_ROOT_PATH / "jaffle_shop",