Merge branch 'main' into release-3.0.0

Author: LeonLuttenberger

.github/workflows/minimum-response-time.yml

@@ -14,6 +14,6 @@ jobs:
         with:
           exempt_user_list: "github-actions[bot]"
           exempt_labels: "help wanted"
-          exempt_authors: "malachi-constant,jaidisido,kukushking,cnfait,dependabot[bot]"
+          exempt_authors: "malachi-constant,jaidisido,kukushking,LeonLuttenberger,cnfait,dependabot[bot]"
           token: ${{secrets.GITHUB_TOKEN}}
           label: needs-triage

awswrangler/_config.py

@@ -48,7 +48,8 @@ class _ConfigArg(NamedTuple):
     "lakeformation_endpoint_url": _ConfigArg(dtype=str, nullable=True, enforced=True, loaded=True),
     "dynamodb_endpoint_url": _ConfigArg(dtype=str, nullable=True, enforced=True, loaded=True),
     "secretsmanager_endpoint_url": _ConfigArg(dtype=str, nullable=True, enforced=True, loaded=True),
-    "timestream_endpoint_url": _ConfigArg(dtype=str, nullable=True, enforced=True, loaded=True),
+    "timestream_query_endpoint_url": _ConfigArg(dtype=str, nullable=True, enforced=True),
+    "timestream_write_endpoint_url": _ConfigArg(dtype=str, nullable=True, enforced=True),
     # Botocore config
     "botocore_config": _ConfigArg(dtype=botocore.config.Config, nullable=True),
     "verify": _ConfigArg(dtype=str, nullable=True, loaded=True),
@@ -70,6 +71,20 @@ class _Config:  # pylint: disable=too-many-instance-attributes,too-many-public-m
     def __init__(self) -> None:
         self._loaded_values: Dict[str, _ConfigValueType] = {}
         name: str
+        self.s3_endpoint_url = None
+        self.athena_endpoint_url = None
+        self.sts_endpoint_url = None
+        self.glue_endpoint_url = None
+        self.redshift_endpoint_url = None
+        self.kms_endpoint_url = None
+        self.emr_endpoint_url = None
+        self.lakeformation_endpoint_url = None
+        self.dynamodb_endpoint_url = None
+        self.secretsmanager_endpoint_url = None
+        self.timestream_write_endpoint_url = None
+        self.timestream_query_endpoint_url = None
+        self.botocore_config = None
+        self.verify = None
         for name in _CONFIG_ARGS:
             self._load_config(name=name)
 
@@ -392,13 +407,22 @@ def secretsmanager_endpoint_url(self, value: Optional[str]) -> None:
         self._set_config_value(key="secretsmanager_endpoint_url", value=value)
 
     @property
-    def timestream_endpoint_url(self) -> Optional[str]:
-        """Property timestream_endpoint_url."""
-        return cast(Optional[str], self["timestream_endpoint_url"])
+    def timestream_query_endpoint_url(self) -> Optional[str]:
+        """Property timestream_query_endpoint_url."""
+        return cast(Optional[str], self["timestream_query_endpoint_url"])
 
-    @timestream_endpoint_url.setter
-    def timestream_endpoint_url(self, value: Optional[str]) -> None:
-        self._set_config_value(key="timestream_endpoint_url", value=value)
+    @timestream_query_endpoint_url.setter
+    def timestream_query_endpoint_url(self, value: Optional[str]) -> None:
+        self._set_config_value(key="timestream_query_endpoint_url", value=value)
+
+    @property
+    def timestream_write_endpoint_url(self) -> Optional[str]:
+        """Property timestream_write_endpoint_url."""
+        return cast(Optional[str], self["timestream_write_endpoint_url"])
+
+    @timestream_write_endpoint_url.setter
+    def timestream_write_endpoint_url(self, value: Optional[str]) -> None:
+        self._set_config_value(key="timestream_write_endpoint_url", value=value)
 
     @property
     def botocore_config(self) -> botocore.config.Config:

awswrangler/_utils.py

@@ -102,8 +102,11 @@ def _get_endpoint_url(service_name: str) -> Optional[str]:
         endpoint_url = _config.config.dynamodb_endpoint_url
     elif service_name == "secretsmanager" and _config.config.secretsmanager_endpoint_url is not None:
         endpoint_url = _config.config.secretsmanager_endpoint_url
-    elif service_name == "timestream" and _config.config.timestream_endpoint_url is not None:
-        endpoint_url = _config.config.timestream_endpoint_url
+    elif service_name == "timestream-write" and _config.config.timestream_write_endpoint_url is not None:
+        endpoint_url = _config.config.timestream_write_endpoint_url
+    elif service_name == "timestream-query" and _config.config.timestream_query_endpoint_url is not None:
+        endpoint_url = _config.config.timestream_query_endpoint_url
+
     return endpoint_url
 
 

awswrangler/redshift.py

@@ -219,6 +219,7 @@ def _validate_parameters(
     distkey: Optional[str],
     sortstyle: str,
     sortkey: Optional[List[str]],
+    primary_keys: Optional[List[str]],
 ) -> None:
     if diststyle not in _RS_DISTSTYLES:
         raise exceptions.InvalidRedshiftDiststyle(f"diststyle must be in {_RS_DISTSTYLES}")
@@ -240,6 +241,14 @@ def _validate_parameters(
                 raise exceptions.InvalidRedshiftSortkey(
                     f"sortkey must be a List of items in the columns list: {cols}. " f"Currently value: {key}"
                 )
+    if primary_keys:
+        if not isinstance(primary_keys, list):
+            raise exceptions.InvalidArgumentType(
+                f"""
+                    primary keys should be of type list[str].
+                    Current value: {primary_keys} is of type {type(primary_keys)}
+                """
+            )
 
 
 def _redshift_types_from_path(
@@ -379,6 +388,7 @@ def _create_table(  # pylint: disable=too-many-locals,too-many-arguments,too-man
         distkey=distkey,
         sortstyle=sortstyle,
         sortkey=sortkey,
+        primary_keys=primary_keys,
     )
     cols_str: str = "".join([f'"{k}" {v},\n' for k, v in redshift_types.items()])[:-2]
     primary_keys_str: str = f",\nPRIMARY KEY ({', '.join(primary_keys)})" if primary_keys else ""

awswrangler/s3/_fs.py

@@ -555,7 +555,7 @@ def open_s3_object(
     newline: Optional[str] = "\n",
     encoding: Optional[str] = "utf-8",
 ) -> Iterator[Union[_S3ObjectBase, io.TextIOWrapper]]:
-    """Return a _S3Object or TextIOWrapper based in the received mode."""
+    """Return a _S3Object or TextIOWrapper based on the received mode."""
     s3obj: Optional[_S3ObjectBase] = None
     text_s3obj: Optional[io.TextIOWrapper] = None
     try:

awswrangler/s3/_read_text.py

@@ -230,7 +230,7 @@ def read_csv(
         The requested parallelism of the read. Only used when `distributed` add-on is installed.
         Parallelism may be limited by the number of files of the dataset. 200 by default.
     pandas_kwargs :
-        KEYWORD arguments forwarded to pandas.read_csv(). You can NOT pass `pandas_kwargs` explicit, just add valid
+        KEYWORD arguments forwarded to pandas.read_csv(). You can NOT pass `pandas_kwargs` explicitly, just add valid
         Pandas arguments in the function call and awswrangler will accept it.
         e.g. wr.s3.read_csv('s3://bucket/prefix/', sep='|', na_values=['null', 'none'], skip_blank_lines=True)
         https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_csv.html
@@ -273,7 +273,7 @@ def read_csv(
     """
     if "pandas_kwargs" in pandas_kwargs:
         raise exceptions.InvalidArgument(
-            "You can NOT pass `pandas_kwargs` explicit, just add valid "
+            "You can NOT pass `pandas_kwargs` explicitly, just add valid "
             "Pandas arguments in the function call and awswrangler will accept it."
             "e.g. wr.s3.read_csv('s3://bucket/prefix/', sep='|', skip_blank_lines=True)"
         )

awswrangler/s3/_read_text_core.py

@@ -18,7 +18,9 @@
 def _get_read_details(path: str, pandas_kwargs: Dict[str, Any]) -> Tuple[str, Optional[str], Optional[str]]:
     if pandas_kwargs.get("compression", "infer") == "infer":
         pandas_kwargs["compression"] = infer_compression(path, compression="infer")
-    mode: str = "r" if pandas_kwargs.get("compression") is None else "rb"
+    mode: str = (
+        "r" if pandas_kwargs.get("compression") is None and pandas_kwargs.get("encoding_errors") != "ignore" else "rb"
+    )
     encoding: Optional[str] = pandas_kwargs.get("encoding", "utf-8")
     newline: Optional[str] = pandas_kwargs.get("lineterminator", None)
     return mode, encoding, newline

poetry.lock

@@ -31,8 +31,10 @@ def wrapper(self, **kwarg):
             assert url == wr.config.glue_endpoint_url
         elif name == "secretsmanager":
             assert url == wr.config.secretsmanager_endpoint_url
-        elif name == "timestream":
-            assert url == wr.config.timestream_endpoint_url
+        elif name == "timestream-write":
+            assert url == wr.config.timestream_write_endpoint_url
+        elif name == "timestream-query":
+            assert url == wr.config.timestream_query_endpoint_url
         return original(self, **kwarg)
 
     with patch("botocore.client.ClientCreator.create_client", new=wrapper):
@@ -117,14 +119,16 @@ def test_basics(path, glue_database, glue_table, workgroup0, workgroup1):
     wr.config.athena_endpoint_url = f"https://athena.{region}.amazonaws.com"
     wr.config.glue_endpoint_url = f"https://glue.{region}.amazonaws.com"
     wr.config.secretsmanager_endpoint_url = f"https://secretsmanager.{region}.amazonaws.com"
-    wr.config.timestream_endpoint_url = f"https://timestream.{region}.amazonaws.com"
+    wr.config.timestream_write_endpoint_url = f"https://ingest.timestream.{region}.amazonaws.com"
+    wr.config.timestream_query_endpoint_url = f"https://query.timestream.{region}.amazonaws.com"
     _urls_test(glue_database)
     os.environ["WR_STS_ENDPOINT_URL"] = f"https://sts.{region}.amazonaws.com"
     os.environ["WR_S3_ENDPOINT_URL"] = f"https://s3.{region}.amazonaws.com"
     os.environ["WR_ATHENA_ENDPOINT_URL"] = f"https://athena.{region}.amazonaws.com"
     os.environ["WR_GLUE_ENDPOINT_URL"] = f"https://glue.{region}.amazonaws.com"
     os.environ["WR_SECRETSMANAGER_ENDPOINT_URL"] = f"https://secretsmanager.{region}.amazonaws.com"
-    os.environ["WR_TIMESTREAM_ENDPOINT_URL"] = f"https://timestream.{region}.amazonaws.com"
+    os.environ["WR_TIMESTREAM_WRITE_ENDPOINT_URL"] = f"https://ingest.timestream.{region}.amazonaws.com"
+    os.environ["WR_TIMESTREAM_QUERY_ENDPOINT_URL"] = f"https://query.timestream.{region}.amazonaws.com"
     wr.config.reset()
     _urls_test(glue_database)
 

tests/unit/test_config.py

@@ -598,3 +598,21 @@ def test_read_parquet_versioned(path) -> None:
         df_temp = wr.s3.read_parquet(path_file, version_id=version_id)
         assert df_temp.equals(df)
         assert version_id == wr.s3.describe_objects(path=path_file, version_id=version_id)[path_file]["VersionId"]
+
+
+def test_read_parquet_schema_validation_with_index_column(path) -> None:
+    path_file = f"{path}file.parquet"
+    df = pd.DataFrame({"idx": [1], "col": [2]})
+    df0 = df.set_index("idx")
+    wr.s3.to_parquet(
+        df=df0,
+        path=path_file,
+        index=True,
+    )
+    df1 = wr.s3.read_parquet(
+        path=path_file,
+        ignore_index=False,
+        columns=["idx", "col"],
+        validate_schema=True,
+    )
+    assert df0.shape == df1.shape