COMPAT/REF: Use s3fs for s3 IO

Author: TomAugspurger

doc/source/whatsnew/v0.20.0.txt

@@ -93,12 +93,20 @@ Backwards incompatible API changes
 
 .. _whatsnew_0200.api:
 
+.. _whatsnew_2000.api.s3
+
+S3 File Handling
+^^^^^^^^^^^^^^^^
+
+pandas now uses `s3fs <http://s3fs.readthedocs.io/>`_ for handling S3 connections. This shouldn't break
+any code, but since s3fs is not a required dependency, you will need to install it separately (like boto
+in prior versions of pandas).
 
 - ``CParserError`` has been renamed to ``ParserError`` in ``pd.read_csv`` and will be removed in the future (:issue:`12665`)
 - ``SparseArray.cumsum()`` and ``SparseSeries.cumsum()`` will now always return ``SparseArray`` and ``SparseSeries`` respectively (:issue:`12855`)
 
 
-
+- ``CParserError`` has been renamed to ``ParserError`` in ``pd.read_csv`` and will be removed in the future (:issue:`12665`)
 
 Other API Changes
 ^^^^^^^^^^^^^^^^^

pandas/io/common.py

@@ -12,6 +12,12 @@
 from pandas.core.common import AbstractMethodError
 from pandas.types.common import is_number
 
+try:
+    from s3fs import S3File
+    need_text_wrapping = (BytesIO, S3File)
+except ImportError:
+    need_text_wrapping = (BytesIO,)
+
 # common NA values
 # no longer excluding inf representations
 # '1.#INF','-1.#INF', '1.#INF000000',
@@ -212,10 +218,10 @@ def get_filepath_or_buffer(filepath_or_buffer, encoding=None,
         return reader, encoding, compression
 
     if _is_s3_url(filepath_or_buffer):
-        from pandas.io.s3 import get_filepath_or_buffer
-        return get_filepath_or_buffer(filepath_or_buffer,
-                                      encoding=encoding,
-                                      compression=compression)
+        from pandas.io import s3
+        return s3.get_filepath_or_buffer(filepath_or_buffer,
+                                         encoding=encoding,
+                                         compression=compression)
 
     # It is a pathlib.Path/py.path.local or string
     filepath_or_buffer = _stringify_path(filepath_or_buffer)
@@ -388,7 +394,7 @@ def _get_handle(path_or_buf, mode, encoding=None, compression=None,
         handles.append(f)
 
     # in Python 3, convert BytesIO or fileobjects passed with an encoding
-    if compat.PY3 and (compression or isinstance(f, compat.BytesIO)):
+    if compat.PY3 and (compression or isinstance(f, need_text_wrapping)):
         from io import TextIOWrapper
         f = TextIOWrapper(f, encoding=encoding)
         handles.append(f)

pandas/io/s3.py

@@ -1,110 +1,33 @@
 """ s3 support for remote file interactivity """
-
-import os
 from pandas import compat
-from pandas.compat import BytesIO
-
 try:
-    import boto
-    from boto.s3 import key
+    import s3fs
 except:
-    raise ImportError("boto is required to handle s3 files")
+    raise ImportError("The s3fs library is required to handle s3 files")
 
 if compat.PY3:
     from urllib.parse import urlparse as parse_url
 else:
     from urlparse import urlparse as parse_url
 
 
-class BotoFileLikeReader(key.Key):
-    """boto Key modified to be more file-like
-
-    This modification of the boto Key will read through a supplied
-    S3 key once, then stop. The unmodified boto Key object will repeatedly
-    cycle through a file in S3: after reaching the end of the file,
-    boto will close the file. Then the next call to `read` or `next` will
-    re-open the file and start reading from the beginning.
-
-    Also adds a `readline` function which will split the returned
-    values by the `\n` character.
-    """
-
-    def __init__(self, *args, **kwargs):
-        encoding = kwargs.pop("encoding", None)  # Python 2 compat
-        super(BotoFileLikeReader, self).__init__(*args, **kwargs)
-        # Add a flag to mark the end of the read.
-        self.finished_read = False
-        self.buffer = ""
-        self.lines = []
-        if encoding is None and compat.PY3:
-            encoding = "utf-8"
-        self.encoding = encoding
-        self.lines = []
-
-    def next(self):
-        return self.readline()
-
-    __next__ = next
-
-    def read(self, *args, **kwargs):
-        if self.finished_read:
-            return b'' if compat.PY3 else ''
-        return super(BotoFileLikeReader, self).read(*args, **kwargs)
-
-    def close(self, *args, **kwargs):
-        self.finished_read = True
-        return super(BotoFileLikeReader, self).close(*args, **kwargs)
-
-    def seekable(self):
-        """Needed for reading by bz2"""
-        return False
-
-    def readline(self):
-        """Split the contents of the Key by '\n' characters."""
-        if self.lines:
-            retval = self.lines[0]
-            self.lines = self.lines[1:]
-            return retval
-        if self.finished_read:
-            if self.buffer:
-                retval, self.buffer = self.buffer, ""
-                return retval
-            else:
-                raise StopIteration
-
-        if self.encoding:
-            self.buffer = "{}{}".format(
-                self.buffer, self.read(8192).decode(self.encoding))
-        else:
-            self.buffer = "{}{}".format(self.buffer, self.read(8192))
-
-        split_buffer = self.buffer.split("\n")
-        self.lines.extend(split_buffer[:-1])
-        self.buffer = split_buffer[-1]
-
-        return self.readline()
+def _strip_schema(url):
+    """Returns the url without the s3:// part"""
+    result = parse_url(url)
+    return result.netloc + result.path
 
 
 def get_filepath_or_buffer(filepath_or_buffer, encoding=None,
                            compression=None):
 
     # Assuming AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_S3_HOST
     # are environment variables
-    parsed_url = parse_url(filepath_or_buffer)
-    s3_host = os.environ.get('AWS_S3_HOST', 's3.amazonaws.com')
-
+    fs = s3fs.S3FileSystem(anon=False)
     try:
-        conn = boto.connect_s3(host=s3_host)
-    except boto.exception.NoAuthHandlerFound:
-        conn = boto.connect_s3(host=s3_host, anon=True)
-
-    b = conn.get_bucket(parsed_url.netloc, validate=False)
-    if compat.PY2 and compression:
-        k = boto.s3.key.Key(b, parsed_url.path)
-        filepath_or_buffer = BytesIO(k.get_contents_as_string(
-            encoding=encoding))
-    else:
-        k = BotoFileLikeReader(b, parsed_url.path, encoding=encoding)
-        k.open('r')  # Expose read errors immediately
-        filepath_or_buffer = k
+        filepath_or_buffer = fs.open(_strip_schema(filepath_or_buffer))
+    except OSError:
+        # boto3 has troubles when trying to access a public file
+        # when credentialed...
+        fs = s3fs.S3FileSystem(anon=True)
+        filepath_or_buffer = fs.open(_strip_schema(filepath_or_buffer))
     return filepath_or_buffer, None, compression

pandas/io/tests/parser/test_network.py

@@ -57,9 +57,9 @@ class TestS3(tm.TestCase):
 
     def setUp(self):
         try:
-            import boto  # noqa
+            import s3fs  # noqa
         except ImportError:
-            raise nose.SkipTest("boto not installed")
+            raise nose.SkipTest("s3fs not installed")
 
     @tm.network
     def test_parse_public_s3_bucket(self):
@@ -193,15 +193,12 @@ def test_parse_public_s3_bucket_nrows_python(self):
 
     @tm.network
     def test_s3_fails(self):
-        import boto
-        with tm.assertRaisesRegexp(boto.exception.S3ResponseError,
-                                   'S3ResponseError: 404 Not Found'):
+        with tm.assertRaises(IOError):
             read_csv('s3://nyqpug/asdf.csv')
 
         # Receive a permission error when trying to read a private bucket.
         # It's irrelevant here that this isn't actually a table.
-        with tm.assertRaisesRegexp(boto.exception.S3ResponseError,
-                                   'S3ResponseError: 403 Forbidden'):
+        with tm.assertRaises(IOError):
             read_csv('s3://cant_get_it/')
 
 if __name__ == '__main__':

pandas/io/tests/test_excel.py

@@ -66,7 +66,7 @@ def _skip_if_no_excelsuite():
 
 def _skip_if_no_boto():
     try:
-        import boto  # NOQA
+        import s3fs  # noqa
     except ImportError:
         raise nose.SkipTest('boto not installed, skipping')