Invalidating cache() for s3fs before reads

Author: igorborgest

awswrangler/pandas.py

@@ -1114,7 +1114,7 @@ def to_redshift(
         :param sortstyle: Sorting can be "COMPOUND" or "INTERLEAVED" (https://docs.aws.amazon.com/redshift/latest/dg/t_Sorting_data.html)
         :param sortkey: List of columns to be sorted
         :param preserve_index: Should we preserve the Dataframe index?
-        :param mode: append or overwrite
+        :param mode: append, overwrite or upsert
         :param cast_columns: Dictionary of columns names and Redshift types to be casted. (E.g. {"col name": "SMALLINT", "col2 name": "FLOAT4"})
         :return: None
         """
@@ -1348,6 +1348,7 @@ def _read_parquet_path(session_primitives: Any,
         procs_cpu_bound = procs_cpu_bound if procs_cpu_bound is not None else session_primitives.procs_cpu_bound if session_primitives.procs_cpu_bound is not None else 1
         use_threads: bool = True if procs_cpu_bound > 1 else False
         fs: S3FileSystem = s3.get_fs(session_primitives=session_primitives)
+        fs.invalidate_cache()
         fs = pa.filesystem._ensure_filesystem(fs)
         logger.debug(f"Reading Parquet table: {path}")
         table = pq.read_table(source=path, columns=columns, filters=filters, filesystem=fs, use_threads=use_threads)

awswrangler/s3.py

@@ -22,6 +22,8 @@ def mkdir_if_not_exists(fs, path):
 
 def get_fs(session_primitives=None):
     aws_access_key_id, aws_secret_access_key, profile_name, config, s3_additional_kwargs = None, None, None, None, None
+    args = {}
+
     if session_primitives:
         if session_primitives.aws_access_key_id:
             aws_access_key_id = session_primitives.aws_access_key_id
@@ -33,17 +35,17 @@ def get_fs(session_primitives=None):
             config = {"retries": {"max_attempts": session_primitives.botocore_max_retries}}
         if session_primitives.s3_additional_kwargs:
             s3_additional_kwargs = session_primitives.s3_additional_kwargs
+
     if profile_name:
-        fs = s3fs.S3FileSystem(profile_name=profile_name,
-                               config_kwargs=config,
-                               s3_additional_kwargs=s3_additional_kwargs)
+        args["profile_name"] = profile_name
     elif aws_access_key_id and aws_secret_access_key:
-        fs = s3fs.S3FileSystem(key=aws_access_key_id,
-                               secret=aws_secret_access_key,
-                               config_kwargs=config,
-                               s3_additional_kwargs=s3_additional_kwargs)
-    else:
-        fs = s3fs.S3FileSystem(config_kwargs=config, s3_additional_kwargs=s3_additional_kwargs)
+        args["key"] = aws_access_key_id,
+        args["secret"] = aws_secret_access_key
+
+    args["config_kwargs"] = config,
+    args["s3_additional_kwargs"] = s3_additional_kwargs
+    fs = s3fs.S3FileSystem(**args)
+    fs.invalidate_cache(path=None)
     return fs