Repository: fsspec/gcsfs (python)

Issue:
_put_file does not allow to specify content encoding
In order to utilize the [decompressive transcoding](https://cloud.google.com/storage/docs/transcoding) of GCS, I need to set the `content_encoding`. But the `_put_file` does not allow to set it. It allows to set `metadata`, but underneath it calls `simple_upload`, where the `content_encoding` must be set in `fixed_key_metadata` optional argument, and this is not exposed to the `_put_file`, so there is no way to set it.

Pull request description:
Expose fixed_key_metadata kwarg for put and pipe
Fixes #543

@racinmat , care to try this?

How would I write a test to assure that this is doing the right thing locally, perhaps with your gzip workflow?

Patch:
diff --git a/gcsfs/core.py b/gcsfs/core.py
index 6290730..28cf61a 100644
--- a/gcsfs/core.py
+++ b/gcsfs/core.py
@@ -284,9 +284,10 @@ class GCSFileSystem(AsyncFileSystem):
         version_aware=False,
         **kwargs,
     ):
+        if cache_timeout:
+            kwargs["listings_expiry_time"] = cache_timeout
         super().__init__(
             self,
-            listings_expiry_time=cache_timeout,
             asynchronous=asynchronous,
             loop=loop,
             **kwargs,
@@ -1105,6 +1106,7 @@ class GCSFileSystem(AsyncFileSystem):
         metadata=None,
         consistency=None,
         content_type="application/octet-stream",
+        fixed_key_metadata=None,
         chunksize=50 * 2**20,
     ):
         # enforce blocksize should be a multiple of 2**18
@@ -1114,10 +1116,24 @@ class GCSFileSystem(AsyncFileSystem):
         out = None
         if size < 5 * 2**20:
             location = await simple_upload(
-                self, bucket, key, data, metadata, consistency, content_type
+                self,
+                bucket,
+                key,
+                data,
+                metadata,
+                consistency,
+                content_type,
+                fixed_key_metadata=fixed_key_metadata,
             )
         else:
-            location = await initiate_upload(self, bucket, key, content_type, metadata)
+            location = await initiate_upload(
+                self,
+                bucket,
+                key,
+                content_type,
+                metadata,
+                fixed_key_metadata=fixed_key_metadata,
+            )
             for offset in range(0, len(data), chunksize):
                 bit = data[offset : offset + chunksize]
                 out = await upload_chunk(
@@ -1140,6 +1156,7 @@ class GCSFileSystem(AsyncFileSystem):
         content_type="application/octet-stream",
         chunksize=50 * 2**20,
         callback=None,
+        fixed_key_metadata=None,
         **kwargs,
     ):
         # enforce blocksize should be a multiple of 2**18
@@ -1165,12 +1182,18 @@ class GCSFileSystem(AsyncFileSystem):
                     consistency=consistency,
                     metadatain=metadata,
                     content_type=content_type,
+                    fixed_key_metadata=fixed_key_metadata,
                 )
                 callback.absolute_update(size)
 
             else:
                 location = await initiate_upload(
-                    self, bucket, key, content_type, metadata
+                    self,
+                    bucket,
+                    key,
+                    content_type,
+                    metadata=metadata,
+                    fixed_key_metadata=fixed_key_metadata,
                 )
                 offset = 0
                 while True:


Test patch:
diff --git a/gcsfs/tests/test_core.py b/gcsfs/tests/test_core.py
index 0d5c4a3..7fce5ec 100644
--- a/gcsfs/tests/test_core.py
+++ b/gcsfs/tests/test_core.py
@@ -1364,3 +1364,10 @@ def test_multiglob(gcs):
     assert gcs.ls(gparent, detail=False) == [f"{root}/t1/t2/t3"]
     gcs.glob(ggparent + "/")
     assert gcs.ls(gparent, detail=False) == [f"{root}/t1/t2/t3"]
+
+
+def test_expiry_keyword():
+    gcs = GCSFileSystem(listings_expiry_time=1, token="anon")
+    assert gcs.dircache.listings_expiry_time == 1
+    gcs = GCSFileSystem(cache_timeout=1, token="anon")
+    assert gcs.dircache.listings_expiry_time == 1