Support for context params in service models

botocore/model.py

@@ -11,7 +11,7 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 """Abstractions to interact with service models."""
-from collections import defaultdict
+from collections import defaultdict, namedtuple
 
 from botocore.compat import OrderedDict
 from botocore.exceptions import (
@@ -84,6 +84,8 @@ class Shape:
         'retryable',
         'document',
         'union',
+        'contextParam',
+        'clientContextParams',
     ]
     MAP_TYPE = OrderedDict
 
@@ -267,6 +269,18 @@ def enum(self):
         return self.metadata.get('enum', [])
 
 
+StaticContextParameter = namedtuple(
+    'StaticContextParameter', ['name', 'value']
+)
+
+
+ContextParameter = namedtuple('ContextParameter', ['name', 'member_name'])
+
+ClientContextParameter = namedtuple(
+    'ClientContextParameter', ['name', 'type', 'documentation']
+)
+
+
 class ServiceModel:
     """
 
@@ -419,6 +433,18 @@ def endpoint_discovery_required(self):
                 return True
         return False
 
+    @CachedProperty
+    def client_context_parameters(self):
+        params = self._service_description.get('clientContextParams', {})
+        return [
+            ClientContextParameter(
+                name=param_name,
+                type=param_val['type'],
+                documentation=param_val['documentation'],
+            )
+            for param_name, param_val in params.items()
+        ]
+
     def _get_metadata_property(self, name):
         try:
             return self.metadata[name]
@@ -428,7 +454,7 @@ def _get_metadata_property(self, name):
             )
 
     # Signature version is one of the rare properties
-    # than can be modified so a CachedProperty is not used here.
+    # that can be modified so a CachedProperty is not used here.
 
     @property
     def signature_version(self):
@@ -560,6 +586,29 @@ def idempotent_members(self):
             and shape.metadata['idempotencyToken']
         ]
 
+    @CachedProperty
+    def static_context_parameters(self):
+        params = self._operation_model.get('staticContextParams', {})
+        return [
+            StaticContextParameter(name=name, value=props.get('value'))
+            for name, props in params.items()
+        ]
+
+    @CachedProperty
+    def context_parameters(self):
+        if not self.input_shape:
+            return []
+
+        return [
+            ContextParameter(
+                name=shape.metadata['contextParam']['name'],
+                member_name=name,
+            )
+            for name, shape in self.input_shape.members.items()
+            if 'contextParam' in shape.metadata
+            and 'name' in shape.metadata['contextParam']
+        ]
+
     @CachedProperty
     def auth_type(self):
         return self._operation_model.get('authtype')

tests/unit/test_model.py

@@ -125,6 +125,43 @@ def test_error_shapes(self):
         self.assertIn('ExceptionOne', error_shape_names)
         self.assertIn('ExceptionTwo', error_shape_names)
 
+    def test_client_context_params(self):
+        service_model = model.ServiceModel(
+            {
+                'metadata': {
+                    'protocol': 'query',
+                    'endpointPrefix': 'endpoint-prefix',
+                    'serviceId': 'MyServiceWithClientContextParams',
+                },
+                'documentation': 'Documentation value',
+                'operations': {},
+                'shapes': {},
+                'clientContextParams': {
+                    'stringClientContextParam': {
+                        'type': 'string',
+                        'documentation': 'str-valued',
+                    },
+                    'booleanClientContextParam': {
+                        'type': 'boolean',
+                        'documentation': 'bool-valued',
+                    },
+                },
+            }
+        )
+        self.assertEqual(len(service_model.client_context_parameters), 2)
+        client_ctx_param1 = service_model.client_context_parameters[0]
+        client_ctx_param2 = service_model.client_context_parameters[1]
+        self.assertEqual(client_ctx_param1.name, 'stringClientContextParam')
+        self.assertEqual(client_ctx_param1.type, 'string')
+        self.assertEqual(client_ctx_param1.documentation, 'str-valued')
+        self.assertEqual(client_ctx_param2.name, 'booleanClientContextParam')
+
+    def test_client_context_params_absent(self):
+        self.assertIsInstance(
+            self.service_model.client_context_parameters, list
+        )
+        self.assertEqual(len(self.service_model.client_context_parameters), 0)
+
 
 class TestOperationModelFromService(unittest.TestCase):
     def setUp(self):
@@ -185,6 +222,25 @@ def setUp(self):
                     'errors': [{'shape': 'NoSuchResourceException'}],
                     'documentation': 'Docs for NoBodyOperation',
                 },
+                'ContextParamOperation': {
+                    'http': {
+                        'method': 'POST',
+                        'requestUri': '/',
+                    },
+                    'name': 'ContextParamOperation',
+                    'input': {'shape': 'ContextParamOperationRequest'},
+                    'output': {'shape': 'OperationNameResponse'},
+                    'errors': [{'shape': 'NoSuchResourceException'}],
+                    'documentation': 'Docs for ContextParamOperation',
+                    'staticContextParams': {
+                        'stringStaticContextParam': {
+                            'value': 'Static Context Param Value',
+                        },
+                        'booleanStaticContextParam': {
+                            'value': True,
+                        },
+                    },
+                },
             },
             'shapes': {
                 'OperationNameRequest': {
@@ -229,6 +285,18 @@ def setUp(self):
                         }
                     },
                 },
+                'ContextParamOperationRequest': {
+                    'type': 'structure',
+                    'members': {
+                        'ContextParamArg': {
+                            'shape': 'stringType',
+                            'contextParam': {
+                                'name': 'contextParamName',
+                            },
+                        }
+                    },
+                    'payload': 'ContextParamArg',
+                },
                 'NoSuchResourceException': {
                     'type': 'structure',
                     'members': {},
@@ -428,6 +496,37 @@ def test_http_checksum_present(self):
             ["crc32", "crc32c", "sha256", "sha1"],
         )
 
+    def test_context_parameter_present(self):
+        service_model = model.ServiceModel(self.model)
+        operation = service_model.operation_model('ContextParamOperation')
+        self.assertEqual(len(operation.context_parameters), 1)
+        context_param = operation.context_parameters[0]
+        self.assertEqual(context_param.name, 'contextParamName')
+        self.assertEqual(context_param.member_name, 'ContextParamArg')
+
+    def test_context_parameter_absent(self):
+        service_model = model.ServiceModel(self.model)
+        operation = service_model.operation_model('OperationTwo')
+        self.assertIsInstance(operation.context_parameters, list)
+        self.assertEqual(len(operation.context_parameters), 0)
+
+    def test_static_context_parameter_present(self):
+        service_model = model.ServiceModel(self.model)
+        operation = service_model.operation_model('ContextParamOperation')
+        self.assertEqual(len(operation.static_context_parameters), 2)
+        static_ctx_param1 = operation.static_context_parameters[0]
+        static_ctx_param2 = operation.static_context_parameters[1]
+        self.assertEqual(static_ctx_param1.name, 'stringStaticContextParam')
+        self.assertEqual(static_ctx_param1.value, 'Static Context Param Value')
+        self.assertEqual(static_ctx_param2.name, 'booleanStaticContextParam')
+        self.assertEqual(static_ctx_param2.value, True)
+
+    def test_static_context_parameter_abent(self):
+        service_model = model.ServiceModel(self.model)
+        operation = service_model.operation_model('OperationTwo')
+        self.assertIsInstance(operation.static_context_parameters, list)
+        self.assertEqual(len(operation.static_context_parameters), 0)
+
 
 class TestOperationModelEventStreamTypes(unittest.TestCase):
     def setUp(self):