Merge pull request #29 from aliyun/con-multipart

For resumable_download, fix #28 so that when tmp file is deleted, we …

CHANGELOG.rst

@@ -3,6 +3,13 @@ OSS SDK for Python 版本记录
 
 Python SDK的版本号遵循 `Semantic Versioning <http://semver.org/>`_ 规则。
 
+Version 2.1.1
+-------------
+
+- 修复：issue #28。
+- 修复：正确的设置连接池大小。
+
+
 Version 2.1.0
 -------------
 

oss2/__init__.py

@@ -1,4 +1,4 @@
-__version__ = '2.1.0'
+__version__ = '2.1.1'
 
 from . import models, exceptions
 

oss2/api.py

@@ -488,6 +488,7 @@ def copy_object(self, source_bucket_name, source_key, target_key, headers=None):
         headers['x-oss-copy-source'] = '/' + source_bucket_name + '/' + source_key
 
         resp = self.__do_object('PUT', target_key, headers=headers)
+
         return PutObjectResult(resp)
 
     def update_object_meta(self, key, headers):
@@ -607,6 +608,7 @@ def complete_multipart_upload(self, key, upload_id, parts, headers=None):
                                 params={'uploadId': upload_id},
                                 data=data,
                                 headers=headers)
+
         return PutObjectResult(resp)
 
     def abort_multipart_upload(self, key, upload_id):
@@ -670,6 +672,7 @@ def upload_part_copy(self, source_bucket_name, source_key, byte_range,
                                 params={'uploadId': target_upload_id,
                                         'partNumber': str(target_part_number)},
                                 headers=headers)
+
         return PutObjectResult(resp)
 
     def list_parts(self, key, upload_id,

oss2/http.py

@@ -28,8 +28,9 @@ class Session(object):
     def __init__(self):
         self.session = requests.Session()
 
-        self.session.mount('http://', requests.adapters.HTTPAdapter(pool_connections=defaults.connection_pool_size))
-        self.session.mount('https://', requests.adapters.HTTPAdapter(pool_connections=defaults.connection_pool_size))
+        psize = defaults.connection_pool_size
+        self.session.mount('http://', requests.adapters.HTTPAdapter(pool_connections=psize, pool_maxsize=psize))
+        self.session.mount('https://', requests.adapters.HTTPAdapter(pool_connections=psize, pool_maxsize=psize))
 
     def do_request(self, req, timeout):
         try:

oss2/resumable.py

@@ -85,7 +85,7 @@ def resumable_download(bucket, key, filename,
     实现的方法是：
         #. 在本地创建一个临时文件，文件名由原始文件名加上一个随机的后缀组成；
         #. 通过指定请求的 `Range` 头按照范围并发读取OSS文件，并写入到临时文件里对应的位置；
-        #. 全部完成之后，把临时文件重名为目标文件 （即 `filename` ）
+        #. 全部完成之后，把临时文件重命名为目标文件 （即 `filename` ）
 
     在上述过程中，断点信息，即已经完成的范围，会保存在磁盘上。因为某种原因下载中断，后续如果下载
     同样的文件，也就是源文件和目标文件一样，就会先读取断点信息，然后只下载缺失的部分。
@@ -296,6 +296,10 @@ def __load_record(self):
             self._del_record()
             record = None
 
+        if record and not os.path.exists(self.filename + record['tmp_suffix']):
+            self._del_record()
+            record = None
+
         if record and self.__is_remote_changed(record):
             utils.silently_remove(self.filename + record['tmp_suffix'])
             self._del_record()

tests/test_download.py

@@ -281,7 +281,7 @@ def test_remote_changed_before_start(self):
         self.__test_insane_record(400, partial(modify_one, key='size', value=1024), old_tmp_exists=False)
         self.__test_insane_record(400, partial(modify_one, key='mtime', value=1024), old_tmp_exists=False)
 
-    def test_remote_changed_during_upload(self):
+    def test_remote_changed_during_download(self):
         oss2.defaults.multiget_threshold = 1
         oss2.defaults.multiget_part_size = 100
         oss2.defaults.multiget_num_threads = 2
@@ -491,6 +491,37 @@ def mock_rename(src, dst):
 
         oss2.utils.silently_remove(abspath)
 
+    def test_tmp_file_removed(self):
+        oss2.defaults.multiget_threshold = 1
+        oss2.defaults.multiget_part_size = 100
+        oss2.defaults.multiget_num_threads = 5
+
+        orig_download_part = oss2.resumable._ResumableDownloader._ResumableDownloader__download_part
+
+        file_size = 123 * 3 + 1
+        key, filename, content = self.__prepare(file_size)
+
+        context = {}
+
+        def mock_download_part(downloader, part, part_number=None):
+            if part.part_number == part_number:
+                r = self.__record(key, filename)
+                context['tmpfile'] = filename + r['tmp_suffix']
+
+                raise RuntimeError("Fail download_part for part: {0}".format(part_number))
+            else:
+                orig_download_part(downloader, part)
+
+        with patch.object(oss2.resumable._ResumableDownloader, '_ResumableDownloader__download_part',
+                          side_effect=partial(mock_download_part, part_number=2),
+                          autospec=True):
+            self.assertRaises(RuntimeError, oss2.resumable_download, self.bucket, key, filename)
+
+        os.remove(context['tmpfile'])
+
+        oss2.resumable_download(self.bucket, key, filename)
+        self.assertFileContent(filename, content)
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_object.py

@@ -3,6 +3,7 @@
 import requests
 import filecmp
 import calendar
+import contextlib
 
 from oss2.exceptions import (ClientError, RequestError,
                              NotFound, NoSuchKey, Conflict, PositionNotEqualToLength, ObjectNotAppendable)