add operation_timeout option.

Author: huiguangjun

alibabacloud_oss_v2/_client.py

@@ -102,12 +102,13 @@ def __init__(
         credentials_provider: Optional[CredentialsProvider] = None,
         http_client: Optional[Union[HttpClient]] = None,
         address_style: Optional[AddressStyle] = None,
-        readwrite_timeout: Optional[int] = None,
+        readwrite_timeout: Optional[Union[int, float]] = None,
         response_handlers: Optional[List] = None,
         response_stream: Optional[bool] = None,
         auth_method: Optional[str] = None,
         feature_flags: Optional[int] = None,
         additional_headers: Optional[List[str]] = None,
+        operation_timeout: Optional[Union[int, float]] = None,
     ) -> None:
         self.product = product
         self.region = region
@@ -124,6 +125,7 @@ def __init__(
         self.auth_method = auth_method
         self.feature_flags = feature_flags or defaults.FF_DEFAULT
         self.additional_headers = additional_headers
+        self.operation_timeout = operation_timeout
 
 
 class _InnerOptions:
@@ -188,6 +190,7 @@ def resolve_operation_kwargs(self, options: _Options, **kwargs):
         options.http_client = kwargs.get("http_client", options.http_client)
         options.readwrite_timeout = kwargs.get("readwrite_timeout", options.readwrite_timeout)
         options.auth_method = kwargs.get("auth_method", options.auth_method)
+        options.operation_timeout = kwargs.get("operation_timeout", None)
 
     def verify_operation(self, op_input: OperationInput, options: _Options) -> None:
         """verify input and options"""
@@ -398,6 +401,11 @@ def _sent_http_request(self, context: SigningContext, options: _Options) -> Http
         retryer = options.retryer
         max_attempts = self.retry_max_attempts(options)
 
+        # operation timeout
+        dealline = None
+        if isinstance(options.operation_timeout, (int, float)):
+            dealline = time.time() +  options.operation_timeout
+
         # Mark body
         marked_body = _MarkedBody(request.body)
         marked_body.mark()
@@ -419,13 +427,21 @@ def _sent_http_request(self, context: SigningContext, options: _Options) -> Http
                 dealy = retryer.retry_delay(tries, error)
                 time.sleep(dealy)
 
+                # operation timeout
+                if dealline is not None and (time.time() > dealline):
+                    break
+
             try:
                 error = None
                 response = self._sent_http_request_once(context, options)
                 break
             except Exception as e:
                 error = e
 
+            # operation timeout
+            if dealline is not None and (time.time() > dealline):
+                break
+
             if marked_body.is_seekable() is False:
                 break
 

tests/unit/test_client.py

@@ -474,6 +474,18 @@ def do_sent_request(op_input: OperationInput, options: Any) -> OperationOutput:
             )
             self.assertIsInstance(self.save_options.http_client, MockHttpClient)
 
+            # "operation_timeout"
+            self.assertEqual(None, clinet._client._options.operation_timeout)
+            clinet.invoke_operation(
+                OperationInput(
+                    op_name='InvokeOperation',
+                    method='GET',
+                    bucket='bucket',
+                ),
+                operation_timeout=40,
+            )
+            self.assertEqual(40, self.save_options.operation_timeout)
+
 
     def test_invoke_operation_user_agent(self):
         self.save_op_context: SigningContext = None
@@ -1162,6 +1174,121 @@ def _do_sent(request: HttpRequest, **kwargs) -> HttpResponse:
             for d in self.save_data:
                 self.assertEqual(data, d)
 
+    def test_invoke_operation_retryable_with_operation_timeout(self):
+        self.save_request: List[HttpRequest] = None
+        self.save_data: List[Any] = None
+
+        def _do_sent(request: HttpRequest, **kwargs) -> HttpResponse:
+            self.save_request.append(request)
+            self.save_data.append(_read_body(request.body))
+            self.assertIsInstance(request.body, io_utils.TeeIterator)
+            raise exceptions.ServiceError(
+                status_code=500,
+                code='InternalError',
+                request_id='id-1234',
+                message='Please contact the server administrator, oss@service.aliyun.com.',
+                ec='',
+                timestamp='',
+                request_target=''
+            )
+
+        cfg = config.Config(
+            region='cn-hangzhou',
+            credentials_provider=credentials.AnonymousCredentialsProvider(),
+        )
+        clinet = client.Client(cfg)
+
+
+        # str, no operation_timeout
+        self.save_request = []
+        self.save_data = []
+        data = 'hello world'
+
+        with mock.patch.object(clinet._client._options.http_client, 'send', new= _do_sent) as _:
+            try:
+                clinet.put_object(models.PutObjectRequest(
+                    bucket='bucket',
+                    key='key',
+                    body=data
+                ))
+                self.fail('should not here')
+            except exceptions.OperationError as err:
+                self.assertIn("InternalError", str(err))
+
+            self.assertEqual(3, len(self.save_request))
+            self.assertEqual(3, len(self.save_data))
+            for d in self.save_data:
+                self.assertEqual(data.encode(), d)
+
+        # str, with 0 operation_timeout
+        self.save_request = []
+        self.save_data = []
+        data = 'hello world'
+
+        with mock.patch.object(clinet._client._options.http_client, 'send', new= _do_sent) as _:
+            try:
+                clinet.put_object(models.PutObjectRequest(
+                    bucket='bucket',
+                    key='key',
+                    body=data
+                ), operation_timeout=0)
+                self.fail('should not here')
+            except exceptions.OperationError as err:
+                self.assertIn("InternalError", str(err))
+
+            self.assertEqual(1, len(self.save_request))
+            self.assertEqual(1, len(self.save_data))
+            for d in self.save_data:
+                self.assertEqual(data.encode(), d)
+
+        # str, with 10s operation_timeout
+        self.save_request = []
+        self.save_data = []
+        data = 'hello world'
+
+        with mock.patch.object(clinet._client._options.http_client, 'send', new= _do_sent) as _:
+            try:
+                clinet.put_object(models.PutObjectRequest(
+                    bucket='bucket',
+                    key='key',
+                    body=data
+                ), operation_timeout=10)
+                self.fail('should not here')
+            except exceptions.OperationError as err:
+                self.assertIn("InternalError", str(err))
+
+            self.assertEqual(3, len(self.save_request))
+            self.assertEqual(3, len(self.save_data))
+            for d in self.save_data:
+                self.assertEqual(data.encode(), d)
+
+        # str, with 4s operation_timeout 2
+        cfg = config.Config(
+            region='cn-hangzhou',
+            credentials_provider=credentials.AnonymousCredentialsProvider(),
+            retryer=retry.StandardRetryer(backoff_delayer=retry.FixedDelayBackoff(2.5))
+        )
+        clinet = client.Client(cfg)
+        self.save_request = []
+        self.save_data = []
+        data = 'hello world'
+
+        with mock.patch.object(clinet._client._options.http_client, 'send', new= _do_sent) as _:
+            try:
+                clinet.put_object(models.PutObjectRequest(
+                    bucket='bucket',
+                    key='key',
+                    body=data
+                ), operation_timeout=4)
+                self.fail('should not here')
+            except exceptions.OperationError as err:
+                self.assertIn("InternalError", str(err))
+
+            self.assertEqual(2, len(self.save_request))
+            self.assertEqual(2, len(self.save_data))
+            for d in self.save_data:
+                self.assertEqual(data.encode(), d)
+
     def test_invoke_operation_addressing_mode(self):
         """ """
         self.save_op_context: SigningContext = None