custom deprecated

Author: YangSen-qn

qiniu/services/storage/legacy.py

@@ -3,15 +3,14 @@
 import os
 import time
 
-from typing_extensions import deprecated
-
 from qiniu import config, http
 from qiniu.auth import Auth
 from qiniu.compat import json
-from qiniu.utils import _file_iter, crc32, rfc_from_timestamp, urlsafe_base64_encode
+from qiniu.utils import _file_iter, crc32, rfc_from_timestamp, urlsafe_base64_encode, deprecated
 
 from qiniu.services.storage.upload_progress_recorder import UploadProgressRecorder
 
+
 @deprecated("use uploader instead")
 class _Resume(object):
     """deprecated 断点续上传类
@@ -87,7 +86,7 @@ def recovery_from_record(self):
                 return 0, None, None
         try:
             if not record['modify_time'] or record['size'] != self.size or \
-                   record['modify_time'] != self.modify_time:
+                    record['modify_time'] != self.modify_time:
                 if self.version == 'v1':
                     return 0
                 elif self.version == 'v2':
@@ -108,8 +107,8 @@ def recovery_from_record(self):
             return record['offset']
         elif self.version == 'v2':
             if not record.__contains__('etags') or len(record['etags']) == 0 or \
-               not record.__contains__('expired_at') or float(record['expired_at']) < time.time() or \
-               not record.__contains__('upload_id'):
+                    not record.__contains__('expired_at') or float(record['expired_at']) < time.time() or \
+                    not record.__contains__('upload_id'):
                 return 0, None, None
             self.blockStatus = record['etags']
             return record['offset'], record['upload_id'], record['expired_at']
@@ -180,7 +179,7 @@ def _upload_v2(self):
         offset, self.uploadId, self.expiredAt = self.recovery_from_record()
         is_resumed = False
         if offset > 0 and self.blockStatus != [] and self.uploadId is not None \
-           and self.expiredAt is not None:
+                and self.expiredAt is not None:
             self.recovery_index = self.blockStatus[-1]['partNumber'] + 1
             is_resumed = True
         else:
@@ -203,7 +202,7 @@ def _upload_v2(self):
                 if config.get_default('default_zone').up_host_backup:
                     host = config.get_default('default_zone').up_host_backup
                 else:
-                    host = config.get_default('default_zone')\
+                    host = config.get_default('default_zone') \
                         .get_up_host_backup_by_token(self.up_token, self.hostscache_dir)
 
             if info.need_retry():

qiniu/services/storage/uploader.py

@@ -1,12 +1,10 @@
 # -*- coding: utf-8 -*-
 import os
 
-from typing_extensions import deprecated
-
 from qiniu.config import _BLOCK_SIZE, get_default
 
 from qiniu.auth import Auth
-from qiniu.utils import crc32, file_crc32, rfc_from_timestamp
+from qiniu.utils import crc32, file_crc32, rfc_from_timestamp, deprecated
 
 from qiniu.services.storage.uploaders import FormUploader, ResumeUploaderV1, ResumeUploaderV2
 from qiniu.services.storage.upload_progress_recorder import UploadProgressRecorder

qiniu/utils.py

@@ -1,4 +1,6 @@
 # -*- coding: utf-8 -*-
+import warnings
+from functools import wraps
 from hashlib import sha1, new as hashlib_new
 from base64 import urlsafe_b64encode, urlsafe_b64decode
 from datetime import datetime, tzinfo, timedelta
@@ -264,3 +266,17 @@ def dt2ts(dt):
         st = (dt - datetime(1970, 1, 1, tzinfo=_UTC_TZINFO())).total_seconds()
 
     return int(st)
+
+
+def deprecated(reason):
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            warnings.warn(
+                "DEPRECATED: {} - {}".format(func.__name__, reason),
+                DeprecationWarning,
+                stacklevel=2
+            )
+            return func(*args, **kwargs)
+        return wrapper
+    return decorator

tests/cases/test_services/test_storage/test_bucket_manager.py

@@ -69,7 +69,7 @@ def test_mkbucketv3(self, bucket_manager, rand_string):
         pass
 
     def test_list_bucket(self, bucket_manager, bucket_name):
-        ret, resp = bucket_manager.list_bucket('z0')
+        ret, resp = bucket_manager.list_bucket('na0')
         assert resp.ok(), resp
         assert any(b.get('tbl') == bucket_name for b in ret)