DNM enable http/2.0 with hyper

examples/object_http20.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+
+import os
+import shutil
+
+import oss2
+
+
+# 以下代码展示了如何启用Http2.0来发送请求。
+
+
+# 首先初始化AccessKeyId、AccessKeySecret、Endpoint等信息。
+# 通过环境变量获取，或者把诸如“<你的AccessKeyId>”替换成真实的AccessKeyId等。
+#
+# 以杭州区域为例，Endpoint可以是：
+#   https://oss-cn-hangzhou.aliyuncs.com
+# 目前Http2.0只支持HTTPS协议访问。
+access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '<你的AccessKeyId>')
+access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '<你的AccessKeySecret>')
+bucket_name = os.getenv('OSS_TEST_BUCKET', '<你的Bucket>')
+endpoint = os.getenv('OSS_TEST_ENDPOINT', '<你的访问域名>')
+
+
+# 确认上面的参数都填写正确了
+for param in (access_key_id, access_key_secret, bucket_name, endpoint):
+    assert '<' not in param, '请设置参数：' + param
+
+# 创建Bucket对象，所有Object相关的接口都可以通过Bucket对象来进行
+bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name, http_version=oss2.HTTP_VERSION_20)
+
+# 上传一段字符串。Object名是motto.txt，内容是一段名言。
+bucket.put_object('motto.txt', 'Never give up. - Jack Ma')
+
+# 下载到本地文件
+bucket.get_object_to_file('motto.txt', '本地文件名.txt')
+
+# 清除本地文件
+os.remove(u'本地文件名.txt')

oss2/__init__.py

@@ -4,7 +4,7 @@
 
 from .api import Service, Bucket, CryptoBucket
 from .auth import Auth, AuthV2, AnonymousAuth, StsAuth, AUTH_VERSION_1, AUTH_VERSION_2, make_auth
-from .http import Session, CaseInsensitiveDict
+from .http import Session, CaseInsensitiveDict, HTTP_VERSION_11, HTTP_VERSION_20
 
 
 from .iterators import (BucketIterator, ObjectIterator,

oss2/api.py

@@ -186,10 +186,10 @@ def progress_callback(bytes_consumed, total_bytes):
 
 class _Base(object):
     def __init__(self, auth, endpoint, is_cname, session, connect_timeout,
-                 app_name='', enable_crc=True):
+                 app_name='', enable_crc=True, http_version=http.HTTP_VERSION_11):
         self.auth = auth
-        self.endpoint = _normalize_endpoint(endpoint.strip())
-        self.session = session or http.Session()
+        self.endpoint = _normalize_endpoint(endpoint.strip(), http_version)
+        self.session = session or http.Session(http_version=http_version)
         self.timeout = defaults.get(connect_timeout, defaults.connect_timeout)
         self.app_name = app_name
         self.enable_crc = enable_crc
@@ -268,11 +268,12 @@ class Service(_Base):
     def __init__(self, auth, endpoint,
                  session=None,
                  connect_timeout=None,
-                 app_name=''):
+                 app_name='',
+                 http_version=http.HTTP_VERSION_11):
         logger.info("Init oss service, endpoint: {0}, connect_timeout: {1}, app_name: {2}".format(
             endpoint, connect_timeout, app_name))
         super(Service, self).__init__(auth, endpoint, False, session, connect_timeout,
-                                      app_name=app_name)
+                                      app_name=app_name, http_version=http_version)
 
     def list_buckets(self, prefix='', marker='', max_keys=100):
         """根据前缀罗列用户的Bucket。
@@ -341,11 +342,12 @@ def __init__(self, auth, endpoint, bucket_name,
                  session=None,
                  connect_timeout=None,
                  app_name='',
-                 enable_crc=True):
+                 enable_crc=True,
+                 http_version=http.HTTP_VERSION_11):
         logger.info("Init oss bucket, endpoint: {0}, isCname: {1}, connect_timeout: {2}, app_name: {3}, enabled_crc: "
                     "{4}".format(endpoint, is_cname, connect_timeout, app_name, enable_crc))
         super(Bucket, self).__init__(auth, endpoint, is_cname, session, connect_timeout,
-                                     app_name, enable_crc)
+                                     app_name, enable_crc, http_version=http_version)
 
         self.bucket_name = bucket_name.strip()
 
@@ -1625,7 +1627,8 @@ def __init__(self, auth, endpoint, bucket_name, crypto_provider,
                  session=None,
                  connect_timeout=None,
                  app_name='',
-                 enable_crc=True):
+                 enable_crc=True,
+                 http_version=http.HTTP_VERSION_11):
 
         if not isinstance(crypto_provider, BaseCryptoProvider):
             raise ClientError('Crypto bucket must provide a valid crypto_provider')
@@ -1634,7 +1637,7 @@ def __init__(self, auth, endpoint, bucket_name, crypto_provider,
         self.bucket_name = bucket_name.strip()
         self.enable_crc = enable_crc
         self.bucket = Bucket(auth, endpoint, bucket_name, is_cname, session, connect_timeout,
-                             app_name, enable_crc=False)
+                             app_name, enable_crc=False, http_version=http_version)
 
     def put_object(self, key, data,
                    headers=None,
@@ -1758,9 +1761,12 @@ def get_object_to_file(self, key, filename,
             return result
 
 
-def _normalize_endpoint(endpoint):
+def _normalize_endpoint(endpoint, http_version):
     if not endpoint.startswith('http://') and not endpoint.startswith('https://'):
-        return 'http://' + endpoint
+        if http_version is http.HTTP_VERSION_20:
+            return 'https://' + endpoint
+        else:
+            return 'http://' + endpoint
     else:
         return endpoint
 

oss2/headers.py

@@ -47,3 +47,6 @@ def set_server_side_encryption(self, algorithm=None, cmk_id=None):
             self[OSS_SERVER_SIDE_ENCRYPTION] = "KMS"
             if cmk_id is not None:
                 self[OSS_SERVER_SIDE_ENCRYPTION_KEY_ID] = cmk_id
+
+    def set_content_length(self, content_length):
+        self["Content-Length"] = content_length

oss2/http.py

@@ -19,26 +19,47 @@
 from .utils import file_object_remaining_bytes, SizedFileAdapter
 
 import logging
+from hyper.contrib import HTTP20Adapter
+import threading
 
 _USER_AGENT = 'aliyun-sdk-python/{0}({1}/{2}/{3};{4})'.format(
     __version__, platform.system(), platform.release(), platform.machine(), platform.python_version())
 
 logger = logging.getLogger(__name__)
 
+HTTP_VERSION_11 = 'http11'
+HTTP_VERSION_20 = 'http20'
+
 class Session(object):
     """属于同一个Session的请求共享一组连接池，如有可能也会重用HTTP连接。"""
-    def __init__(self):
+    def __init__(self, http_version=HTTP_VERSION_11):
         self.session = requests.Session()
 
-        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))
+        self.http_version = http_version
+        self.__lock = threading.Lock()
+
+        if http_version is HTTP_VERSION_20:
+            self.session.mount('https://', HTTP20Adapter())
+        else:
+            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:
-            logger.debug("Send request, method: {0}, url: {1}, params: {2}, headers: {3}, timeout: {4}".format(
-                req.method, req.url, req.params, req.headers, timeout))
-            return Response(self.session.request(req.method, req.url,
+            logger.debug("Send request, method: {0}, url: {1}, params: {2}, headers: {3}, timeout: {4}, http_version: {5}".format(
+                req.method, req.url, req.params, req.headers, timeout, self.http_version))
+            if self.http_version is HTTP_VERSION_20:
+                with self.__lock:
+                    resp = Response(self.session.request(req.method, req.url,
+                                                 data=req.data,
+                                                 params=req.params,
+                                                 headers=req.headers,
+                                                 stream=True,
+                                                 timeout=timeout))
+                    return resp
+            else:
+                return Response(self.session.request(req.method, req.url,
                                                  data=req.data,
                                                  params=req.params,
                                                  headers=req.headers,

scripts/patch_hyper.py

@@ -0,0 +1,55 @@
+#!/usr/bin/python
+
+import os
+import hyper
+import h2
+
+def fix_hyper():
+    if hyper.__version__ == "0.7.0":
+        hyper_dir = os.path.dirname(hyper.__file__)
+        fix_file_path = hyper_dir + "/common/headers.py"
+
+        f_read = open(fix_file_path,'r+')
+        flist = f_read.readlines()
+        if flist[244] == """    SPECIAL_SNOWFLAKES = set([b'set-cookie', b'set-cookie2'])\n""":
+            flist[244] = """    SPECIAL_SNOWFLAKES = set([b'set-cookie', b'set-cookie2', b'date', b'if-modified-since', b'if-unmodified-since', b'authorization'])\n"""
+
+            print " ====================================================================================="
+            print " # OSS already patch to fix hyper library "
+            print " # fixed file name: ", fix_file_path
+            print " # fixed line number: 244"
+            print " # More detail to see: https://github.com/Lukasa/hyper/issues/314 "
+            print " ====================================================================================="
+        f_read.close()
+
+        f_wte = open(fix_file_path, 'w+')
+        f_wte.writelines(flist)
+        f_wte.close()
+
+def fix_h2():
+    if h2.__version__ == "2.6.2":
+        h2_dir = os.path.dirname(h2.__file__)
+        fix_file_path = h2_dir + "/stream.py"
+
+        f_read = open(fix_file_path, 'r+')
+        flist = f_read.readlines()
+        if flist[337] == """        raise StreamClosedError(self.stream_id)\n""":
+            flist[337] = """        #raise StreamClosedError(self.stream_id)\n        return []\n"""
+            print " ====================================================================================="
+            print " # OSS already patch to fix h2 library "
+            print " # fixed file name: ", fix_file_path
+            print " # fixed line number: 337"
+            print " ====================================================================================="
+        f_read.close()
+
+        f_wte = open(fix_file_path, 'w+')
+        f_wte.writelines(flist)
+        f_wte.close()
+
+def main():
+    fix_hyper()
+    fix_h2()
+
+
+if __name__ == "__main__":
+    main()

setup.py

@@ -29,6 +29,7 @@
     packages=['oss2'],
     install_requires=['requests!=2.9.0',
                       'crcmod>=1.7',
+                      'hyper>=0.7.0',
                       'pycryptodome>=3.4.7',
                       'aliyun-python-sdk-kms>=2.4.1',
                       'aliyun-python-sdk-core>=2.6.2' if sys.version_info[0] == 2 else 'aliyun-python-sdk-core-v3>=2.5.5'],

tests/common.py

@@ -26,6 +26,7 @@
 OSS_STS_ARN = os.getenv("OSS_TEST_STS_ARN")
 
 OSS_AUTH_VERSION = None
+OSS_HTTP_VERSION = None
 
 def random_string(n):
     return ''.join(random.choice(string.ascii_lowercase) for i in range(n))
@@ -80,21 +81,23 @@ def setUp(self):
         oss2.defaults.multiget_part_size = self.default_multiget_part_size
         oss2.defaults.multiget_num_threads = random.randint(1, 5)
 
-        global OSS_AUTH_VERSION
+        global OSS_AUTH_VERSION, OSS_HTTP_VERSION
         OSS_AUTH_VERSION = os.getenv('OSS_TEST_AUTH_VERSION')
+        OSS_HTTP_VERSION = os.getenv('OSS_TEST_HTTP_VERSION')
 
-        self.bucket = oss2.Bucket(oss2.make_auth(OSS_ID, OSS_SECRET, OSS_AUTH_VERSION), OSS_ENDPOINT, OSS_BUCKET)
+        self.bucket = oss2.Bucket(oss2.make_auth(OSS_ID, OSS_SECRET, OSS_AUTH_VERSION), OSS_ENDPOINT, OSS_BUCKET, http_version=OSS_HTTP_VERSION)
 
         try:
             self.bucket.create_bucket()
         except:
             pass
 
         self.rsa_crypto_bucket = oss2.CryptoBucket(oss2.make_auth(OSS_ID, OSS_SECRET, OSS_AUTH_VERSION), OSS_ENDPOINT, OSS_BUCKET,
-                                             crypto_provider=oss2.LocalRsaProvider())
+                                             crypto_provider=oss2.LocalRsaProvider(), http_version=OSS_HTTP_VERSION)
 
+        # Special handle for http20, Because now KMS don't support http20, kms_crypto_bucket create with http11
         self.kms_crypto_bucket = oss2.CryptoBucket(oss2.make_auth(OSS_ID, OSS_SECRET, OSS_AUTH_VERSION), OSS_ENDPOINT, OSS_BUCKET,
-                                             crypto_provider=oss2.AliKMSProvider(OSS_ID, OSS_SECRET, OSS_REGION, OSS_CMK))
+                                             crypto_provider=oss2.AliKMSProvider(OSS_ID, OSS_SECRET, OSS_REGION, OSS_CMK), http_version=oss2.HTTP_VERSION_11)
 
         self.key_list = []
         self.temp_files = []

tests/test_bucket.py

@@ -541,6 +541,26 @@ def test_xml_input_output(self):
             self.assertEqual(result.allow_empty_referer, True)
             self.assertEqual(result.referers[0], to_string(u'阿里云'))
 
+class TestHttp20OverBucket(TestBucket):
+    """
+        当环境变量使用oss2.HTTP11时，则重新设置为HTTP20, 再运行TestBucket，反之亦然
+    """
+    def __init__(self, *args, **kwargs):
+        super(TestHttp20OverBucket, self).__init__(*args, **kwargs)
+
+    def setUp(self):
+        if os.getenv('OSS_TEST_HTTP_VERSION') == oss2.HTTP_VERSION_11:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_20
+        else:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_11
+        super(TestHttp20OverBucket, self).setUp()
+
+    def tearDown(self):
+        if os.getenv('OSS_TEST_HTTP_VERSION') == oss2.HTTP_VERSION_11:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_20
+        else:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_11
+        super(TestHttp20OverBucket, self).tearDown()
 
 if __name__ == '__main__':
     unittest.main()

tests/test_chinese.py

@@ -63,7 +63,27 @@ def test_get_symlink(self):
 
         result = self.bucket.get_symlink(symlink)
         self.assertEqual(result.target_key, key)
-
+
+class TestHttp20OverChinese(TestChinese):
+    """
+        当环境变量使用oss2.HTTP11时，则重新设置为HTTP20, 再运行TestChinese，反之亦然
+    """
+    def __init__(self, *args, **kwargs):
+        super(TestHttp20OverChinese, self).__init__(*args, **kwargs)
+
+    def setUp(self):
+        if os.getenv('OSS_TEST_HTTP_VERSION') == oss2.HTTP_VERSION_11:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_20
+        else:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_11
+        super(TestHttp20OverChinese, self).setUp()
+
+    def tearDown(self):
+        if os.getenv('OSS_TEST_HTTP_VERSION') == oss2.HTTP_VERSION_11:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_20
+        else:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_11
+        super(TestHttp20OverChinese, self).tearDown()
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()

tests/test_download.py

@@ -609,6 +609,27 @@ def test_resumable_incomplete_download(self):
             except:
                 self.assertTrue(False)
 
+class TestHttp20OverDownload(TestDownload):
+    """
+        当环境变量使用oss2.HTTP11时，则重新设置为HTTP20, 再运行TestDownload，反之亦然
+    """
+    def __init__(self, *args, **kwargs):
+        super(TestHttp20OverDownload, self).__init__(*args, **kwargs)
+
+    def setUp(self):
+        if os.getenv('OSS_TEST_HTTP_VERSION') == oss2.HTTP_VERSION_11:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_20
+        else:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_11
+        super(TestHttp20OverDownload, self).setUp()
+
+    def tearDown(self):
+        if os.getenv('OSS_TEST_HTTP_VERSION') == oss2.HTTP_VERSION_11:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_20
+        else:
+            os.environ['OSS_TEST_HTTP_VERSION'] = oss2.HTTP_VERSION_11
+        super(TestHttp20OverDownload, self).tearDown()
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_headers.py

@@ -23,6 +23,14 @@ def test_check_requestHeader(self):
         self.assertTrue("x-oss-server-side-encryption" not in myHeader)
         self.assertTrue("x-oss-server-side-encryption-key-id" not in myHeader)
 
+    def test_set_content_length(self):
+        myHeader = RequestHeader()
+
+        myHeader.set_content_length("100")
+        self.assertTrue(myHeader["Content-Length"] is "100")
+
+        myHeader.set_content_length("200")
+        self.assertTrue(myHeader["Content-Length"] is "200")
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()