remove deprecated pw.io.s3_csv.read (#8976)

GitOrigin-RevId: 9c8dea7fcc8a025497968a38e1e55538624e9f61

Author: zxqfd555

CHANGELOG.md

@@ -12,6 +12,10 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Changed
 - **BREAKING** The Elasticsearch and BigQuery connectors have been moved to the Scale license tier. You can obtain the Scale tier license for free at https://pathway.com/get-license.
 - **BREAKING** `pw.io.fs.read` no longer accepts `format="raw"`. Use `format="binary"` to read binary objects, `format="plaintext_by_file"` to read plaintext objects per file, or `format="plaintext"` to read plaintext objects split into lines.
+- **BREAKING** The `pw.io.s3_csv.read` connector has been removed. Please use `pw.io.s3.read` with `format="csv"` instead.
+
+### Fixed
+- `pw.io.s3.read` and `pw.io.s3.write` now also check the `AWS_PROFILE` environment variable for AWS credentials if none are explicitly provided.
 
 ## [0.24.1] - 2025-07-03
 

integration_tests/s3/test_s3_generic.py

@@ -24,10 +24,11 @@ def test_s3_backfilling(snapshot_access, tmp_path: pathlib.Path, s3_path: str):
 
     input_contents = "key,value\n1,Hello\n2,World"
     put_aws_object(s3_input_path, input_contents)
-    table = pw.io.s3_csv.read(
+    table = pw.io.s3.read(
         s3_path,
         aws_s3_settings=get_aws_s3_settings(),
         schema=pw.schema_from_types(key=int, value=str),
+        format="csv",
         mode="static",
         autocommit_duration_ms=1000,
         name="1",
@@ -44,11 +45,12 @@ def test_s3_backfilling(snapshot_access, tmp_path: pathlib.Path, s3_path: str):
 
     input_contents = "key,value\n1,Hello\n2,World\n3,Bonjour\n4,Monde\n"
     put_aws_object(s3_input_path, input_contents)
-    table = pw.io.s3_csv.read(
+    table = pw.io.s3.read(
         s3_path,
         aws_s3_settings=get_aws_s3_settings(),
         schema=pw.schema_from_types(key=int, value=str),
         mode="static",
+        format="csv",
         autocommit_duration_ms=1000,
         name="1",
     )
@@ -68,11 +70,12 @@ def test_s3_backfilling(snapshot_access, tmp_path: pathlib.Path, s3_path: str):
     output_path = tmp_path / "output_final.csv"
     put_aws_object(s3_input_path, input_contents)
     put_aws_object(s3_input_path_2, input_contents_2)
-    table = pw.io.s3_csv.read(
+    table = pw.io.s3.read(
         s3_path,
         aws_s3_settings=get_aws_s3_settings(),
         schema=pw.schema_from_types(key=int, value=str),
         mode="static",
+        format="csv",
         autocommit_duration_ms=1000,
         name="1",
     )
@@ -189,15 +192,16 @@ def test_s3_alternative_path(tmp_path: pathlib.Path, s3_path: str):
     put_aws_object(input_s3_path, input_contents)
     write_lines(model_output_path, input_contents)
 
-    table = pw.io.s3_csv.read(
+    table = pw.io.s3.read(
         f"s3://aws-integrationtest/{input_s3_path}",
-        aws_s3_settings=pw.io.s3_csv.AwsS3Settings(
+        aws_s3_settings=pw.io.s3.AwsS3Settings(
             access_key=os.environ["AWS_S3_ACCESS_KEY"],
             secret_access_key=os.environ["AWS_S3_SECRET_ACCESS_KEY"],
             region="eu-central-1",
         ),
         schema=pw.schema_from_types(key=int, value=str),
         mode="static",
+        format="csv",
         autocommit_duration_ms=1000,
     )
 
@@ -217,13 +221,14 @@ def test_s3_wrong_path(tmp_path: pathlib.Path, s3_path: str):
     input_s3_path = f"{s3_path}/input.csv"
     output_path = tmp_path / "output.csv"
 
-    table = pw.io.s3_csv.read(
+    table = pw.io.s3.read(
         f"s3://aws-integrationtest/{input_s3_path}",
-        aws_s3_settings=pw.io.s3_csv.AwsS3Settings(
+        aws_s3_settings=pw.io.s3.AwsS3Settings(
             access_key=os.environ["AWS_S3_ACCESS_KEY"],
             secret_access_key=os.environ["AWS_S3_SECRET_ACCESS_KEY"],
             region="eu-central-1",
         ),
+        format="csv",
         schema=pw.schema_from_types(key=int, value=str),
         mode="static",
         autocommit_duration_ms=1000,
@@ -248,10 +253,11 @@ def test_s3_creds_from_profiles(tmp_path: pathlib.Path, s3_path: str, monkeypatc
     put_aws_object(input_s3_path, input_contents)
     write_lines(model_output_path, input_contents)
 
-    table = pw.io.s3_csv.read(
+    table = pw.io.s3.read(
         f"s3://aws-integrationtest/{input_s3_path}",
-        aws_s3_settings=pw.io.s3_csv.AwsS3Settings(region="eu-central-1"),
+        aws_s3_settings=pw.io.s3.AwsS3Settings(region="eu-central-1"),
         schema=pw.schema_from_types(key=int, value=str),
+        format="csv",
         mode="static",
         autocommit_duration_ms=1000,
     )

python/pathway/io/__init__.py

@@ -27,7 +27,6 @@
     questdb,
     redpanda,
     s3,
-    s3_csv,
     slack,
     sqlite,
 )
@@ -59,7 +58,6 @@
     "slack",
     "subscribe",
     "s3",
-    "s3_csv",
     "gdrive",
     "sqlite",
     "pubsub",

python/pathway/io/s3_csv/__init__.py

@@ -437,7 +437,6 @@ def test_traceback_static():
         pw.io.plaintext.read,
         pw.io.python.read,
         pw.io.redpanda.read,
-        pw.io.s3_csv.read,
         pw.io.csv.write,
         pw.io.fs.write,
         pw.io.http.write,

python/pathway/tests/test_error_messages.py

@@ -3992,7 +3992,15 @@ impl AwsS3Settings {
             } else {
                 let aws_credentials = AwsCredentials::from_sts_env("aws-creds")
                     .or_else(|_| AwsCredentials::from_env())
-                    .or_else(|_| AwsCredentials::from_profile(self.profile.as_deref()))
+                    // The rust-s3 crate doesn't check the AWS_PROFILE env var, so we do it by ourselves
+                    .or_else(|_| {
+                        AwsCredentials::from_profile(
+                            self.profile
+                                .clone()
+                                .or_else(|| std::env::var("AWS_PROFILE").ok())
+                                .as_deref(),
+                        )
+                    })
                     .or_else(|_| AwsCredentials::from_instance_metadata());
 
                 // first, try to deduce credentials from various sources