Adding support to download be stream of data

coreapi/__init__.py

@@ -4,7 +4,7 @@
 from coreapi.document import Array, Document, Link, Object, Error, Field
 
 
-__version__ = '2.3.3'
+__version__ = '2.3.4'
 __all__ = [
     'Array', 'Document', 'Link', 'Object', 'Error', 'Field',
     'Client',

coreapi/client.py

@@ -140,7 +140,7 @@ def reload(self, document, format=None, force_codec=False):
         return self.get(document.url, format=format, force_codec=force_codec)
 
     def action(self, document, keys, params=None, validate=True, overrides=None,
-               action=None, encoding=None, transform=None):
+               action=None, encoding=None, transform=None, stream=False):
         if (action is not None) or (encoding is not None) or (transform is not None):
             # Fallback for v1.x overrides.
             # Will be removed at some point, most likely in a 2.1 release.
@@ -175,4 +175,4 @@ def action(self, document, keys, params=None, validate=True, overrides=None,
 
         # Perform the action, and return a new document.
         transport = determine_transport(self.transports, link.url)
-        return transport.transition(link, self.decoders, params=params, link_ancestors=link_ancestors)
+        return transport.transition(link, self.decoders, params=params, link_ancestors=link_ancestors, stream=stream)

coreapi/codecs/base.py

@@ -19,9 +19,14 @@ def dump(self, *args, **kwargs):
         # Fallback for v1.x interface
         return self.encode(*args, **kwargs)
 
-    def load(self, *args, **kwargs):
+    def support_streaming(self):
+        return False
+
+    def load(self, chunks, *args, **kwargs):
         # Fallback for v1.x interface
-        return self.decode(*args, **kwargs)
+        if not(self.support_streaming()):
+            chunks = bytes().join(chunks) or bytes()
+        return self.decode(chunks, *args, **kwargs)
 
     @property
     def supports(self):

coreapi/codecs/download.py

@@ -6,6 +6,7 @@
 import os
 import posixpath
 import tempfile
+from tqdm import tqdm
 
 
 def _unique_output_path(path):
@@ -102,26 +103,34 @@ class DownloadCodec(BaseCodec):
     media_type = '*/*'
     format = 'download'
 
-    def __init__(self, download_dir=None):
+    def __init__(self, download_dir=None, progress_bar=False):
         """
         `download_dir` - The path to use for file downloads.
         """
         self._delete_on_close = download_dir is None
         self._download_dir = download_dir
+        self._progress_bar = progress_bar
+
+    def support_streaming(self):
+        return True
 
     @property
     def download_dir(self):
         return self._download_dir
 
-    def decode(self, bytestring, **options):
+    def decode(self, chunks, **options):
         base_url = options.get('base_url')
         content_type = options.get('content_type')
         content_disposition = options.get('content_disposition')
 
         # Write the download to a temporary .download file.
-        fd, temp_path = tempfile.mkstemp(suffix='.download')
+        fd, temp_path = tempfile.mkstemp(suffix='.download', dir=self._download_dir)
         file_handle = os.fdopen(fd, 'wb')
-        file_handle.write(bytestring)
+        content_length = options.get("content-length", None)
+        if content_length and self._progress_bar:
+            chunks = tqdm(chunks, total=content_length, unit="mb")
+        for chunk in chunks:
+            file_handle.write(chunk)
         file_handle.close()
 
         # Determine the output filename.

coreapi/transports/http.py

@@ -86,7 +86,14 @@ def _get_method(action):
     return action.upper()
 
 
-def _get_encoding(encoding):
+def _get_encoding(encoding, params):
+    has_file = False
+    if params is not None:
+        for value in params.values():
+            if hasattr(value, 'read'):
+                has_file = True
+    if has_file:
+        return 'multipart/form-data'
     if not encoding:
         return 'application/json'
     return encoding
@@ -226,7 +233,6 @@ def _build_http_request(session, url, method, headers=None, encoding=None, param
                 opts['data'] = params.data
             upload_headers = _get_upload_headers(params.data)
             opts['headers'].update(upload_headers)
-
     request = requests.Request(method, url, **opts)
     return session.prepare_request(request)
 
@@ -268,14 +274,18 @@ def _coerce_to_error(obj, default_title):
         return Error(title=default_title, content={'messages': obj})
     elif obj is None:
         return Error(title=default_title)
+    elif hasattr(obj, "read"):
+        return Error(title=default_title, content={'messages': obj.read().decode("utf-8")})
     return Error(title=default_title, content={'message': obj})
 
 
 def _decode_result(response, decoders, force_codec=False):
     """
     Given an HTTP response, return the decoded Core API document.
     """
-    if response.content:
+    chunk_size = 1024 * 1024
+    chunks = response.iter_content(chunk_size)
+    if chunks:
         # Content returned in response. We should decode it.
         if force_codec:
             codec = decoders[0]
@@ -290,8 +300,9 @@ def _decode_result(response, decoders, force_codec=False):
             options['content_type'] = response.headers['content-type']
         if 'content-disposition' in response.headers:
             options['content_disposition'] = response.headers['content-disposition']
-
-        result = codec.load(response.content, **options)
+        if 'content-length' in response.headers:
+            options["content-length"] = int(response.headers["content-length"]) / chunk_size
+        result = codec.load(chunks, **options)
     else:
         # No content returned in response.
         result = None
@@ -366,19 +377,24 @@ def __init__(self, credentials=None, headers=None, auth=None, session=None, requ
     def headers(self):
         return self._headers
 
-    def transition(self, link, decoders, params=None, link_ancestors=None, force_codec=False):
+    def transition(self, link, decoders, params=None, link_ancestors=None, force_codec=False, stream=False):
         session = self._session
         method = _get_method(link.action)
-        encoding = _get_encoding(link.encoding)
+        encoding = _get_encoding(link.encoding, params)
         params = _get_params(method, encoding, link.fields, params)
         url = _get_url(link.url, params.path)
         headers = _get_headers(url, decoders)
         headers.update(self.headers)
 
         request = _build_http_request(session, url, method, headers, encoding, params)
+
         settings = session.merge_environment_settings(request.url, None, None, None, None)
+        settings["stream"] = stream
         response = session.send(request, **settings)
-        result = _decode_result(response, decoders, force_codec)
+        result = None
+        if response.status_code != 204:  # no content
+            result = _decode_result(response, decoders, force_codec)
+        response.close()
 
         if isinstance(result, Document) and link_ancestors:
             result = _handle_inplace_replacements(result, link, link_ancestors)

requirements.txt

@@ -3,6 +3,7 @@ coreschema
 itypes
 requests
 uritemplate
+tqdm
 
 # Testing requirements
 coverage

setup.py

@@ -66,7 +66,8 @@ def get_package_data(package):
         'coreschema',
         'requests',
         'itypes',
-        'uritemplate'
+        'uritemplate',
+        'tqdm'
     ],
     entry_points={
         'coreapi.codecs': [

tests/test_integration.py

@@ -23,6 +23,14 @@ def __init__(self, content):
         self.url = 'http://example.org'
         self.status_code = 200
 
+    def iter_content(self, *args, **kwargs):
+        n = 2
+        list_of_chunks = list(self.content[i:i + n] for i in range(0, len(self.content), n))
+        return list_of_chunks
+
+    def close(self):
+        return
+
 
 # Basic integration tests.
 

tests/test_transitions.py

@@ -8,7 +8,7 @@
 class MockTransport(HTTPTransport):
     schemes = ['mock']
 
-    def transition(self, link, decoders, params=None, link_ancestors=None):
+    def transition(self, link, decoders, params=None, link_ancestors=None, stream=True):
         if link.action == 'get':
             document = Document(title='new', content={'new': 123})
         elif link.action in ('put', 'post'):

tests/test_transport.py

@@ -26,9 +26,16 @@ def __init__(self, content):
         self.url = 'http://example.org'
         self.status_code = 200
 
+    def iter_content(self, *args, **kwargs):
+        n = 2
+        list_of_chunks = list(self.content[i:i + n] for i in range(0, len(self.content), n))
+        return list_of_chunks
 
+    def close(self):
+        return
 # Test transport errors.
 
+
 def test_unknown_scheme():
     with pytest.raises(NetworkError):
         determine_transport(transports, 'ftp://example.org')