two version of R2R are hereHEADmaster

1 files changed, 384 insertions, 0 deletions

diff --git a/.venv/lib/python3.12/site-packages/botocore/validate.py b/.venv/lib/python3.12/site-packages/botocore/validate.py
new file mode 100644
index 00000000..82aabd66
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/botocore/validate.py
@@ -0,0 +1,384 @@
+"""User input parameter validation.
+
+This module handles user input parameter validation
+against a provided input model.
+
+Note that the objects in this module do *not* mutate any
+arguments.  No type version happens here.  It is up to another
+layer to properly convert arguments to any required types.
+
+Validation Errors
+-----------------
+
+
+"""
+
+import decimal
+import json
+from datetime import datetime
+
+from botocore.exceptions import ParamValidationError
+from botocore.utils import is_json_value_header, parse_to_aware_datetime
+
+
+def validate_parameters(params, shape):
+    """Validates input parameters against a schema.
+
+    This is a convenience function that validates parameters against a schema.
+    You can also instantiate and use the ParamValidator class directly if you
+    want more control.
+
+    If there are any validation errors then a ParamValidationError
+    will be raised.  If there are no validation errors than no exception
+    is raised and a value of None is returned.
+
+    :param params: The user provided input parameters.
+
+    :type shape: botocore.model.Shape
+    :param shape: The schema which the input parameters should
+        adhere to.
+
+    :raise: ParamValidationError
+
+    """
+    validator = ParamValidator()
+    report = validator.validate(params, shape)
+    if report.has_errors():
+        raise ParamValidationError(report=report.generate_report())
+
+
+def type_check(valid_types):
+    def _create_type_check_guard(func):
+        def _on_passes_type_check(self, param, shape, errors, name):
+            if _type_check(param, errors, name):
+                return func(self, param, shape, errors, name)
+
+        def _type_check(param, errors, name):
+            if not isinstance(param, valid_types):
+                valid_type_names = [str(t) for t in valid_types]
+                errors.report(
+                    name,
+                    'invalid type',
+                    param=param,
+                    valid_types=valid_type_names,
+                )
+                return False
+            return True
+
+        return _on_passes_type_check
+
+    return _create_type_check_guard
+
+
+def range_check(name, value, shape, error_type, errors):
+    failed = False
+    min_allowed = float('-inf')
+    if 'min' in shape.metadata:
+        min_allowed = shape.metadata['min']
+        if value < min_allowed:
+            failed = True
+    elif hasattr(shape, 'serialization'):
+        # Members that can be bound to the host have an implicit min of 1
+        if shape.serialization.get('hostLabel'):
+            min_allowed = 1
+            if value < min_allowed:
+                failed = True
+    if failed:
+        errors.report(name, error_type, param=value, min_allowed=min_allowed)
+
+
+class ValidationErrors:
+    def __init__(self):
+        self._errors = []
+
+    def has_errors(self):
+        if self._errors:
+            return True
+        return False
+
+    def generate_report(self):
+        error_messages = []
+        for error in self._errors:
+            error_messages.append(self._format_error(error))
+        return '\n'.join(error_messages)
+
+    def _format_error(self, error):
+        error_type, name, additional = error
+        name = self._get_name(name)
+        if error_type == 'missing required field':
+            return (
+                f"Missing required parameter in {name}: "
+                f"\"{additional['required_name']}\""
+            )
+        elif error_type == 'unknown field':
+            unknown_param = additional['unknown_param']
+            valid_names = ', '.join(additional['valid_names'])
+            return (
+                f'Unknown parameter in {name}: "{unknown_param}", '
+                f'must be one of: {valid_names}'
+            )
+        elif error_type == 'invalid type':
+            param = additional['param']
+            param_type = type(param)
+            valid_types = ', '.join(additional['valid_types'])
+            return (
+                f'Invalid type for parameter {name}, value: {param}, '
+                f'type: {param_type}, valid types: {valid_types}'
+            )
+        elif error_type == 'invalid range':
+            param = additional['param']
+            min_allowed = additional['min_allowed']
+            return (
+                f'Invalid value for parameter {name}, value: {param}, '
+                f'valid min value: {min_allowed}'
+            )
+        elif error_type == 'invalid length':
+            param = additional['param']
+            min_allowed = additional['min_allowed']
+            return (
+                f'Invalid length for parameter {name}, value: {param}, '
+                f'valid min length: {min_allowed}'
+            )
+        elif error_type == 'unable to encode to json':
+            return 'Invalid parameter {} must be json serializable: {}'.format(
+                name,
+                additional['type_error'],
+            )
+        elif error_type == 'invalid type for document':
+            param = additional['param']
+            param_type = type(param)
+            valid_types = ', '.join(additional['valid_types'])
+            return (
+                f'Invalid type for document parameter {name}, value: {param}, '
+                f'type: {param_type}, valid types: {valid_types}'
+            )
+        elif error_type == 'more than one input':
+            members = ', '.join(additional['members'])
+            return (
+                f'Invalid number of parameters set for tagged union structure '
+                f'{name}. Can only set one of the following keys: '
+                f'{members}.'
+            )
+        elif error_type == 'empty input':
+            members = ', '.join(additional['members'])
+            return (
+                f'Must set one of the following keys for tagged union'
+                f'structure {name}: {members}.'
+            )
+
+    def _get_name(self, name):
+        if not name:
+            return 'input'
+        elif name.startswith('.'):
+            return name[1:]
+        else:
+            return name
+
+    def report(self, name, reason, **kwargs):
+        self._errors.append((reason, name, kwargs))
+
+
+class ParamValidator:
+    """Validates parameters against a shape model."""
+
+    def validate(self, params, shape):
+        """Validate parameters against a shape model.
+
+        This method will validate the parameters against a provided shape model.
+        All errors will be collected before returning to the caller.  This means
+        that this method will not stop at the first error, it will return all
+        possible errors.
+
+        :param params: User provided dict of parameters
+        :param shape: A shape model describing the expected input.
+
+        :return: A list of errors.
+
+        """
+        errors = ValidationErrors()
+        self._validate(params, shape, errors, name='')
+        return errors
+
+    def _check_special_validation_cases(self, shape):
+        if is_json_value_header(shape):
+            return self._validate_jsonvalue_string
+        if shape.type_name == 'structure' and shape.is_document_type:
+            return self._validate_document
+
+    def _validate(self, params, shape, errors, name):
+        special_validator = self._check_special_validation_cases(shape)
+        if special_validator:
+            special_validator(params, shape, errors, name)
+        else:
+            getattr(self, f'_validate_{shape.type_name}')(
+                params, shape, errors, name
+            )
+
+    def _validate_jsonvalue_string(self, params, shape, errors, name):
+        # Check to see if a value marked as a jsonvalue can be dumped to
+        # a json string.
+        try:
+            json.dumps(params)
+        except (ValueError, TypeError) as e:
+            errors.report(name, 'unable to encode to json', type_error=e)
+
+    def _validate_document(self, params, shape, errors, name):
+        if params is None:
+            return
+
+        if isinstance(params, dict):
+            for key in params:
+                self._validate_document(params[key], shape, errors, key)
+        elif isinstance(params, list):
+            for index, entity in enumerate(params):
+                self._validate_document(
+                    entity, shape, errors, '%s[%d]' % (name, index)
+                )
+        elif not isinstance(params, ((str,), int, bool, float)):
+            valid_types = (str, int, bool, float, list, dict)
+            valid_type_names = [str(t) for t in valid_types]
+            errors.report(
+                name,
+                'invalid type for document',
+                param=params,
+                param_type=type(params),
+                valid_types=valid_type_names,
+            )
+
+    @type_check(valid_types=(dict,))
+    def _validate_structure(self, params, shape, errors, name):
+        if shape.is_tagged_union:
+            if len(params) == 0:
+                errors.report(name, 'empty input', members=shape.members)
+            elif len(params) > 1:
+                errors.report(
+                    name, 'more than one input', members=shape.members
+                )
+
+        # Validate required fields.
+        for required_member in shape.metadata.get('required', []):
+            if required_member not in params:
+                errors.report(
+                    name,
+                    'missing required field',
+                    required_name=required_member,
+                    user_params=params,
+                )
+        members = shape.members
+        known_params = []
+        # Validate known params.
+        for param in params:
+            if param not in members:
+                errors.report(
+                    name,
+                    'unknown field',
+                    unknown_param=param,
+                    valid_names=list(members),
+                )
+            else:
+                known_params.append(param)
+        # Validate structure members.
+        for param in known_params:
+            self._validate(
+                params[param],
+                shape.members[param],
+                errors,
+                f'{name}.{param}',
+            )
+
+    @type_check(valid_types=(str,))
+    def _validate_string(self, param, shape, errors, name):
+        # Validate range.  For a string, the min/max constraints
+        # are of the string length.
+        # Looks like:
+        # "WorkflowId":{
+        #   "type":"string",
+        #   "min":1,
+        #   "max":256
+        #  }
+        range_check(name, len(param), shape, 'invalid length', errors)
+
+    @type_check(valid_types=(list, tuple))
+    def _validate_list(self, param, shape, errors, name):
+        member_shape = shape.member
+        range_check(name, len(param), shape, 'invalid length', errors)
+        for i, item in enumerate(param):
+            self._validate(item, member_shape, errors, f'{name}[{i}]')
+
+    @type_check(valid_types=(dict,))
+    def _validate_map(self, param, shape, errors, name):
+        key_shape = shape.key
+        value_shape = shape.value
+        for key, value in param.items():
+            self._validate(key, key_shape, errors, f"{name} (key: {key})")
+            self._validate(value, value_shape, errors, f'{name}.{key}')
+
+    @type_check(valid_types=(int,))
+    def _validate_integer(self, param, shape, errors, name):
+        range_check(name, param, shape, 'invalid range', errors)
+
+    def _validate_blob(self, param, shape, errors, name):
+        if isinstance(param, (bytes, bytearray, str)):
+            return
+        elif hasattr(param, 'read'):
+            # File like objects are also allowed for blob types.
+            return
+        else:
+            errors.report(
+                name,
+                'invalid type',
+                param=param,
+                valid_types=[str(bytes), str(bytearray), 'file-like object'],
+            )
+
+    @type_check(valid_types=(bool,))
+    def _validate_boolean(self, param, shape, errors, name):
+        pass
+
+    @type_check(valid_types=(float, decimal.Decimal) + (int,))
+    def _validate_double(self, param, shape, errors, name):
+        range_check(name, param, shape, 'invalid range', errors)
+
+    _validate_float = _validate_double
+
+    @type_check(valid_types=(int,))
+    def _validate_long(self, param, shape, errors, name):
+        range_check(name, param, shape, 'invalid range', errors)
+
+    def _validate_timestamp(self, param, shape, errors, name):
+        # We don't use @type_check because datetimes are a bit
+        # more flexible.  You can either provide a datetime
+        # object, or a string that parses to a datetime.
+        is_valid_type = self._type_check_datetime(param)
+        if not is_valid_type:
+            valid_type_names = [str(datetime), 'timestamp-string']
+            errors.report(
+                name, 'invalid type', param=param, valid_types=valid_type_names
+            )
+
+    def _type_check_datetime(self, value):
+        try:
+            parse_to_aware_datetime(value)
+            return True
+        except (TypeError, ValueError, AttributeError):
+            # Yes, dateutil can sometimes raise an AttributeError
+            # when parsing timestamps.
+            return False
+
+
+class ParamValidationDecorator:
+    def __init__(self, param_validator, serializer):
+        self._param_validator = param_validator
+        self._serializer = serializer
+
+    def serialize_to_request(self, parameters, operation_model):
+        input_shape = operation_model.input_shape
+        if input_shape is not None:
+            report = self._param_validator.validate(
+                parameters, operation_model.input_shape
+            )
+            if report.has_errors():
+                raise ParamValidationError(report=report.generate_report())
+        return self._serializer.serialize_to_request(
+            parameters, operation_model
+        )