ENH: add fsspec support

doc/source/getting_started/install.rst

@@ -267,8 +267,9 @@ SQLAlchemy                1.1.4              SQL support for databases other tha
 SciPy                     0.19.0             Miscellaneous statistical functions
 XLsxWriter                0.9.8              Excel writing
 blosc                                        Compression for HDF5
+fsspec                    0.7.4              File operations handling
 fastparquet               0.3.2              Parquet reading / writing
-gcsfs                     0.2.2              Google Cloud Storage access
+gcsfs                     0.6.0              Google Cloud Storage access
 html5lib                                     HTML parser for read_html (see :ref:`note <optional_html>`)
 lxml                      3.8.0              HTML parser for read_html (see :ref:`note <optional_html>`)
 matplotlib                2.2.2              Visualization
@@ -282,7 +283,7 @@ pyreadstat                                   SPSS files (.sav) reading
 pytables                  3.4.3              HDF5 reading / writing
 pyxlsb                    1.0.6              Reading for xlsb files
 qtpy                                         Clipboard I/O
-s3fs                      0.3.0              Amazon S3 access
+s3fs                      0.4.0              Amazon S3 access
 tabulate                  0.8.3              Printing in Markdown-friendly format (see `tabulate`_)
 xarray                    0.8.2              pandas-like API for N-dimensional data
 xclip                                        Clipboard I/O on linux

doc/source/whatsnew/v1.1.0.rst

@@ -197,6 +197,20 @@ For a full example, see: :ref:`timeseries.adjust-the-start-of-the-bins`.
 
 .. _whatsnew_110.enhancements.other:
 
+fsspec now used for filesystem handling
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+For reading and writing to filesystems other than local and reading from HTTP(S),
+the optional dependency ``fsspec`` will be used to dispatch operations. This will give unchanged
+functionality for S3 and GCS storage, which were already supported, but also add
+support for several other storage implementations such as Azure Datalake and Blob,
+SSH, FTP, dropbox and github. For docs and capabilities, see the `fsspec docs`_.
+
+In the future, we will implement a way to pass parameters to the invoked
+filesystem instances.
+
+.. _fsspec docs: https://filesystem-spec.readthedocs.io/en/latest/
+
 Other enhancements
 ^^^^^^^^^^^^^^^^^^
 

environment.yml

@@ -98,7 +98,8 @@ dependencies:
 
   - pyqt>=5.9.2  # pandas.read_clipboard
   - pytables>=3.4.3  # pandas.read_hdf, DataFrame.to_hdf
-  - s3fs  # pandas.read_csv... when using 's3://...' path
+  - s3fs>=0.4.0  # pandas.read_csv... when using 's3://...' path
+  - fsspec>=0.7.4  # for generic remote file operations
   - sqlalchemy  # pandas.read_sql, DataFrame.to_sql
   - xarray  # DataFrame.to_xarray
   - cftime  # Needed for downstream xarray.CFTimeIndex test

pandas/compat/_optional.py

@@ -8,8 +8,9 @@
 VERSIONS = {
     "bs4": "4.6.0",
     "bottleneck": "1.2.1",
+    "fsspec": "0.7.4",
     "fastparquet": "0.3.2",
-    "gcsfs": "0.2.2",
+    "gcsfs": "0.6.0",
     "lxml.etree": "3.8.0",
     "matplotlib": "2.2.2",
     "numexpr": "2.6.2",
@@ -20,7 +21,7 @@
     "pytables": "3.4.3",
     "pytest": "5.0.1",
     "pyxlsb": "1.0.6",
-    "s3fs": "0.3.0",
+    "s3fs": "0.4.0",
     "scipy": "1.2.0",
     "sqlalchemy": "1.1.4",
     "tables": "3.4.3",

pandas/io/common.py

@@ -31,6 +31,7 @@
 
 from pandas._typing import FilePathOrBuffer
 from pandas.compat import _get_lzma_file, _import_lzma
+from pandas.compat._optional import import_optional_dependency
 
 from pandas.core.dtypes.common import is_file_like
 
@@ -126,20 +127,6 @@ def stringify_path(
     return _expand_user(filepath_or_buffer)
 
 
-def is_s3_url(url) -> bool:
-    """Check for an s3, s3n, or s3a url"""
-    if not isinstance(url, str):
-        return False
-    return parse_url(url).scheme in ["s3", "s3n", "s3a"]
-
-
-def is_gcs_url(url) -> bool:
-    """Check for a gcs url"""
-    if not isinstance(url, str):
-        return False
-    return parse_url(url).scheme in ["gcs", "gs"]
-
-
 def urlopen(*args, **kwargs):
     """
     Lazy-import wrapper for stdlib urlopen, as that imports a big chunk of
@@ -150,38 +137,20 @@ def urlopen(*args, **kwargs):
     return urllib.request.urlopen(*args, **kwargs)
 
 
-def get_fs_for_path(filepath: str):
+def is_fsspec_url(url: FilePathOrBuffer) -> bool:
     """
-    Get appropriate filesystem given a filepath.
-    Supports s3fs, gcs and local file system.
-
-    Parameters
-    ----------
-    filepath : str
-        File path. e.g s3://bucket/object, /local/path, gcs://pandas/obj
-
-    Returns
-    -------
-    s3fs.S3FileSystem, gcsfs.GCSFileSystem, None
-        Appropriate FileSystem to use. None for local filesystem.
+    Returns true if fsspec is installed and the given URL looks like
+    something fsspec can handle
     """
-    if is_s3_url(filepath):
-        from pandas.io import s3
-
-        return s3.get_fs()
-    elif is_gcs_url(filepath):
-        from pandas.io import gcs
-
-        return gcs.get_fs()
-    else:
-        return None
+    return isinstance(url, str) and ("::" in url or "://" in url)
 
 
 def get_filepath_or_buffer(
     filepath_or_buffer: FilePathOrBuffer,
     encoding: Optional[str] = None,
     compression: Optional[str] = None,
     mode: Optional[str] = None,
+    **storage_options: Dict[str, Any],
 ):
     """
     If the filepath_or_buffer is a url, translate and return the buffer.
@@ -194,6 +163,8 @@ def get_filepath_or_buffer(
     compression : {{'gzip', 'bz2', 'zip', 'xz', None}}, optional
     encoding : the encoding to use to decode bytes, default is 'utf-8'
     mode : str, optional
+    storage_options: dict
+        passed on to fsspec, if using it; this is not yet accessed by the public API
 
     Returns
     -------
@@ -204,6 +175,7 @@ def get_filepath_or_buffer(
     filepath_or_buffer = stringify_path(filepath_or_buffer)
 
     if isinstance(filepath_or_buffer, str) and is_url(filepath_or_buffer):
+        # TODO: fsspec can also handle HTTP via requests, but leaving this unchanged
         req = urlopen(filepath_or_buffer)
         content_encoding = req.headers.get("Content-Encoding", None)
         if content_encoding == "gzip":
@@ -213,19 +185,13 @@ def get_filepath_or_buffer(
         req.close()
         return reader, encoding, compression, True
 
-    if is_s3_url(filepath_or_buffer):
-        from pandas.io import s3
-
-        return s3.get_filepath_or_buffer(
-            filepath_or_buffer, encoding=encoding, compression=compression, mode=mode
-        )
+    if is_fsspec_url(filepath_or_buffer):
+        fsspec = import_optional_dependency("fsspec")
 
-    if is_gcs_url(filepath_or_buffer):
-        from pandas.io import gcs
-
-        return gcs.get_filepath_or_buffer(
-            filepath_or_buffer, encoding=encoding, compression=compression, mode=mode
-        )
+        file_obj = fsspec.open(
+            filepath_or_buffer, mode=mode or "rb", **storage_options
+        ).open()
+        return file_obj, encoding, compression, True
 
     if isinstance(filepath_or_buffer, (str, bytes, mmap.mmap)):
         return _expand_user(filepath_or_buffer), None, compression, False

pandas/io/parquet.py

@@ -8,12 +8,7 @@
 
 from pandas import DataFrame, get_option
 
-from pandas.io.common import (
-    get_filepath_or_buffer,
-    get_fs_for_path,
-    is_gcs_url,
-    is_s3_url,
-)
+from pandas.io.common import get_filepath_or_buffer, is_fsspec_url
 
 
 def get_engine(engine: str) -> "BaseImpl":
@@ -107,6 +102,14 @@ def write(
         # write_to_dataset does not support a file-like object when
         # a directory path is used, so just pass the path string.
         if partition_cols is not None:
+            # user may provide filesystem= with an instance, in which case it takes
+            # priority and fsspec need not analyse the path
+            if is_fsspec_url(path) and "filesystem" not in kwargs:
+                import_optional_dependency("fsspec")
+                import fsspec.core
+
+                fs, path = fsspec.core.url_to_fs(path)
+                kwargs["filesystem"] = fs
             self.api.parquet.write_to_dataset(
                 table,
                 path,
@@ -122,9 +125,17 @@ def write(
             file_obj_or_path.close()
 
     def read(self, path, columns=None, **kwargs):
-        parquet_ds = self.api.parquet.ParquetDataset(
-            path, filesystem=get_fs_for_path(path), **kwargs
-        )
+        if is_fsspec_url(path) and "filesystem" not in kwargs:
+            import_optional_dependency("fsspec")
+            import fsspec.core
+
+            fs, path = fsspec.core.url_to_fs(path)
+            parquet_ds = self.api.parquet.ParquetDataset(path, filesystem=fs, **kwargs)
+        else:
+            parquet_ds = self.api.parquet.ParquetDataset(path, **kwargs)
+            # this key valid for ParquetDataset but not read_pandas
+            kwargs.pop("filesystem", None)
+
         kwargs["columns"] = columns
         result = parquet_ds.read_pandas(**kwargs).to_pandas()
         return result
@@ -164,13 +175,11 @@ def write(
         if partition_cols is not None:
             kwargs["file_scheme"] = "hive"
 
-        if is_s3_url(path) or is_gcs_url(path):
-            # if path is s3:// or gs:// we need to open the file in 'wb' mode.
-            # TODO: Support 'ab'
+        if is_fsspec_url(path):
+            fsspec = import_optional_dependency("fsspec")
 
-            path, _, _, _ = get_filepath_or_buffer(path, mode="wb")
-            # And pass the opened file to the fastparquet internal impl.
-            kwargs["open_with"] = lambda path, _: path
+            # if filesystem is provided by fsspec, file must be opened in 'wb' mode.
+            kwargs["open_with"] = lambda path, _: fsspec.open(path, "wb").open()
         else:
             path, _, _, _ = get_filepath_or_buffer(path)
 
@@ -185,17 +194,11 @@ def write(
             )
 
     def read(self, path, columns=None, **kwargs):
-        if is_s3_url(path):
-            from pandas.io.s3 import get_file_and_filesystem
+        if is_fsspec_url(path):
+            fsspec = import_optional_dependency("fsspec")
 
-            # When path is s3:// an S3File is returned.
-            # We need to retain the original path(str) while also
-            # pass the S3File().open function to fastparquet impl.
-            s3, filesystem = get_file_and_filesystem(path)
-            try:
-                parquet_file = self.api.ParquetFile(path, open_with=filesystem.open)
-            finally:
-                s3.close()
+            open_with = lambda path, _: fsspec.open(path, "rb").open()
+            parquet_file = self.api.ParquetFile(path, open_with=open_with)
         else:
             path, _, _, _ = get_filepath_or_buffer(path)
             parquet_file = self.api.ParquetFile(path)

pandas/tests/io/test_common.py

@@ -367,3 +367,9 @@ def test_unknown_engine(self):
             df.to_csv(path)
             with pytest.raises(ValueError, match="Unknown engine"):
                 pd.read_csv(path, engine="pyt")
+
+
+def test_is_fsspec_url():
+    assert icom.is_fsspec_url("gcs://pandas/somethingelse.com")
+    assert icom.is_fsspec_url("gs://pandas/somethingelse.com")
+    assert not icom.is_fsspec_url("random:pandas/somethingelse.com")