KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use to_json_schema method in pandera

Best Python code snippet using pandera_python

restapi.py

Source:restapi.py

...122    @abc.abstractmethod123    def to_openapi_responses(self, service, spec) -> dict:124        return {}125    @abc.abstractmethod126    def to_json_schema(self, service, spec, direction) -> dict:127        return {}128class BinaryData(RestMethodParam):129    def __init__(self, mediatypes="*/*", required=False):130        if not isinstance(mediatypes, list):131            mediatypes = [mediatypes]132        self._mediatypes = mediatypes133        self._required = required134    def to_json_schema(self, service, spec, direction):135        return {136            "type": "string",137            "format": "binary",138            "required": self._required,139        }140    @property141    def _binary_content_schema(self):142        return {143            mediatype: {"schema": self.to_json_schema(None, None, None)}144            for mediatype in self._mediatypes145        }146    def to_openapi_requestbody(self, service, spec):147        return {"content": self._binary_content_schema}148    def to_openapi_query_parameters(self, service, spec):149        raise NotImplementedError(150            "BinaryData are not (?yet?) supported as query paramters"151        )152    def to_openapi_responses(self, service, spec):153        return {"200": {"content": self._binary_content_schema}}154    def to_response(self, service, result):155        if not isinstance(result, http.Response):156            # The response has not been build by the called method...157            result = self._to_http_response(result)158        return result159    def from_params(self, service, params):160        return params161    def _to_http_response(self, result):162        mediatype = self._mediatypes[0] if len(self._mediatypes) == 1 else "*/*"163        headers = [164            ("Content-Type", mediatype),165            ("X-Content-Type-Options", "nosniff"),166            ("Content-Disposition", http.content_disposition("file")),167            ("Content-Length", len(result)),168        ]169        return http.request.make_response(result, headers)170class CerberusValidator(RestMethodParam):171    def __init__(self, schema):172        """173        :param schema: can be dict as cerberus schema, an instance of174                       cerberus.Validator or a sting with the method name to175                       call on the service to get the schema or the validator176        """177        self._schema = schema178    def from_params(self, service, params):179        validator = self.get_cerberus_validator(service, "input")180        if validator.validate(params):181            return validator.document182        raise UserError(_("BadRequest %s") % validator.errors)183    def to_response(self, service, result):184        validator = self.get_cerberus_validator(service, "output")185        if validator.validate(result):186            return validator.document187        raise SystemError(_("Invalid Response %s") % validator.errors)188    def to_openapi_query_parameters(self, service, spec):189        json_schema = self.to_json_schema(service, spec, "input")190        parameters = []191        for prop, spec in list(json_schema["properties"].items()):192            params = {193                "name": prop,194                "in": "query",195                "required": prop in json_schema["required"],196                "allowEmptyValue": spec.get("nullable", False),197                "default": spec.get("default"),198            }199            if spec.get("schema"):200                params["schema"] = spec.get("schema")201            else:202                params["schema"] = {"type": spec["type"]}203            if spec.get("items"):204                params["schema"]["items"] = spec.get("items")205            if "enum" in spec:206                params["schema"]["enum"] = spec["enum"]207            parameters.append(params)208            if spec["type"] == "array":209                # To correctly handle array into the url query string,210                # the name must ends with []211                params["name"] = params["name"] + "[]"212        return parameters213    def to_openapi_requestbody(self, service, spec):214        json_schema = self.to_json_schema(service, spec, "input")215        return {"content": {"application/json": {"schema": json_schema}}}216    def to_openapi_responses(self, service, spec):217        json_schema = self.to_json_schema(service, spec, "output")218        return {"200": {"content": {"application/json": {"schema": json_schema}}}}219    def get_cerberus_validator(self, service, direction):220        assert direction in ("input", "output")221        schema = self._schema222        if isinstance(self._schema, str):223            validator_component = service.component(usage="cerberus.validator")224            schema = validator_component.get_validator_handler(225                service, self._schema, direction226            )()227        if isinstance(schema, Validator):228            return schema229        if isinstance(schema, dict):230            return Validator(schema, purge_unknown=True)231        raise Exception(_("Unable to get cerberus schema from %s") % self._schema)232    def to_json_schema(self, service, spec, direction):233        schema = self.get_cerberus_validator(service, direction).schema234        return cerberus_to_json(schema)235class CerberusListValidator(CerberusValidator):236    def __init__(self, schema, min_items=None, max_items=None, unique_items=None):237        """238        :param schema: Cerberus list item schema239                       can be dict as cerberus schema, an instance of240                       cerberus.Validator or a sting with the method name to241                       call on the service to get the schema or the validator242        :param min_items: A list instance is valid against "min_items" if its243                          size is greater than, or equal to, min_items.244                          The value MUST be a non-negative integer.245        :param max_items: A list instance is valid against "max_items" if its246                          size is less than, or equal to, max_items.247                          The value MUST be a non-negative integer.248        :param unique_items: Used to document that the list should only249                             contain unique items.250                             (Not enforced at validation time)251        """252        super(CerberusListValidator, self).__init__(schema=schema)253        self._min_items = min_items254        self._max_items = max_items255        self._unique_items = unique_items256    def from_params(self, service, params):257        return self._do_validate(service, data=params, direction="input")258    def to_response(self, service, result):259        return self._do_validate(service, data=result, direction="output")260    def to_openapi_query_parameters(self, service, spec):261        raise NotImplementedError("List are not (?yet?) supported as query paramters")262    # pylint: disable=W8120,W8115263    def _do_validate(self, service, data, direction):264        validator = self.get_cerberus_validator(service, direction)265        values = []266        ExceptionClass = UserError if direction == "input" else SystemError267        for idx, p in enumerate(data):268            if not validator.validate(p):269                raise ExceptionClass(270                    _(271                        "BadRequest item %(idx)s :%(errors)s",272                        idx=idx,273                        errors=validator.errors,274                    )275                )276            values.append(validator.document)277        if self._min_items is not None and len(values) < self._min_items:278            raise ExceptionClass(279                _(280                    "BadRequest: Not enough items in the list (%(current)s < %(expected)s)",281                    current=len(values),282                    expected=self._min_items,283                )284            )285        if self._max_items is not None and len(values) > self._max_items:286            raise ExceptionClass(287                _(288                    "BadRequest: Too many items in the list (%(current)s > %(expected)s)",289                    current=len(values),290                    expected=self._max_items,291                )292            )293        return values294    def to_json_schema(self, service, spec, direction):295        cerberus_schema = self.get_cerberus_validator(service, direction).schema296        json_schema = cerberus_to_json(cerberus_schema)297        json_schema = {"type": "array", "items": json_schema}298        if self._min_items is not None:299            json_schema["minItems"] = self._min_items300        if self._max_items is not None:301            json_schema["maxItems"] = self._max_items302        if self._unique_items is not None:303            json_schema["uniqueItems"] = self._unique_items304        return json_schema305class MultipartFormData(RestMethodParam):306    def __init__(self, parts):307        """This allows to create multipart/form-data endpoints.308        :param parts:  list of RestMethodParam309        """310        if not isinstance(parts, dict):311            raise ValidationError(_("You must provide a dict of RestMethodParam"))312        self._parts = parts313    def to_openapi_properties(self, service, spec, direction):314        properties = {}315        for key, part in self._parts.items():316            properties[key] = part.to_json_schema(service, spec, direction)317        return properties318    def to_openapi_encoding(self):319        encodings = {}320        for key, part in self._parts.items():321            if isinstance(part, BinaryData):322                encodings[key] = {"contentType": ", ".join(part._mediatypes)}323        return encodings324    def to_json_schema(self, service, spec, direction):325        res = {326            "multipart/form-data": {327                "schema": {328                    "type": "object",329                    "properties": self.to_openapi_properties(service, spec, direction),330                }331            }332        }333        encoding = self.to_openapi_encoding()334        if len(encoding) > 0:335            res["multipart/form-data"]["schema"]["encoding"] = encoding336        return res337    def from_params(self, service, params):338        for key, part in self._parts.items():339            param = None340            if isinstance(part, BinaryData):341                param = part.from_params(service, params[key])342            else:343                # If the part is not Binary, it should be JSON344                try:345                    json_param = json.loads(346                        params[key]347                    )  # multipart ony sends its parts as string348                except json.JSONDecodeError as error:349                    raise ValidationError(350                        _(351                            "%(key)'s JSON content is malformed: %(error)s",352                            key=key,353                            error=error,354                        )355                    ) from error356                param = part.from_params(service, json_param)357            params[key] = param358        return params359    def to_openapi_query_parameters(self, service, spec):360        raise NotImplementedError(361            "MultipartFormData are not (?yet?) supported as query paramters"362        )363    def to_openapi_requestbody(self, service, spec):364        return {"content": self.to_json_schema(service, spec, "input")}365    def to_openapi_responses(self, service, spec):366        return {"200": {"content": self.to_json_schema(service, spec, "output")}}367    def to_response(self, service, result):...

typing.py

Source:typing.py

...33        min_value = item[1] if len(item) > 1 else None34        max_value = item[2] if len(item) > 2 else None35        if base_type not in (int, float):36            raise ValueError('Range can be used only with int and float types')37        def to_json_schema(cls):38            sch = convert_schema(cls.inner)39            if cls.min_value is not None:40                sch['minimum'] = cls.min_value41            if cls.max_value is not None:42                sch['maximum'] = cls.max_value43            return sch44        return type(45            'Range[{0}, {1}, {2}]'.format(base_type.__name__, min_value, max_value),46            (BaseObject,), {47                '__json_schema__': classmethod(to_json_schema),48                'inner': base_type,49                'min_value': min_value,50                'max_value': max_value51            }52        )53class PatternFactory(object):54    def __getitem__(self, pat):55        def to_json_schema(cls):56            return {57                'type': 'string',58                'pattern': cls.pattern59            }60        return type(61            'Pattern["{0}"]'.format(pat),62            (BaseObject,), {63                '__json_schema__': classmethod(to_json_schema),64                'pattern': pat65            }66        )67class DefaultFactory(object):68    def __getitem__(self, item):69        if len(item) != 2:70            raise ValueError('Default requires base type and a default value')71        base_type, default_value = item72        if not isinstance(base_type, type):73            raise ValueError('Default base_type must be a type')74        def to_json_schema(cls):75            return {76                'type': cls.base_type.to_json_schema(),77                'pattern': cls.pattern78            }79        return type(80            'Default[{0}, {1!r}]'.format(base_type.__name__, default_value),81            (BaseObject,), {82                '__json_schema__': classmethod(to_json_schema),83                'base_type': base_type,84                'default_value': default_value85            }86        )87Range = RangeFactory()88Pattern = PatternFactory()...

test_json_schema.py

Source:test_json_schema.py

...6request_samples = read_recordings_as_dict()7object_sample = json.loads(request_samples[0]["response"]["body"])8test_objects = [[{"fork": True}], [{}]]9def test_to_schema_from_array():10    schema = to_json_schema(object_sample, UpdateMode.GEN, schema=None)11    assert_that(schema, has_key("$schema"))12    assert_that(13        schema, has_entry("$schema", starts_with("http://json-schema.org/schema#"))14    )15    assert_that(schema, has_entry("type", "array"))16    assert_that(schema, has_entry("items", instance_of(dict)))17def test_schema_with_array():18    schema = to_json_schema(test_objects[0], UpdateMode.GEN, schema=None)19    assert_that(schema, has_entry("items", instance_of(dict)))20    items = schema["items"]21    assert isinstance(items, dict)  # typeguard22    assert_that(items, has_entry("required", ["fork"]))23    assert_that(items, has_entry("properties", instance_of(dict)))24    properties = items["properties"]25    assert_that(properties, has_entries(fork=instance_of(dict)))26    forks = properties["fork"]27    assert_that(forks, has_entries(type="boolean"))28def test_updating_schema_removes_required():29    schema = to_json_schema(test_objects[0], UpdateMode.GEN, schema=None)30    schema = to_json_schema(test_objects[1], UpdateMode.GEN, schema=schema)31    items = schema["items"]32    assert isinstance(items, dict)  # typeguard33    assert_that(items, any_of(is_not(has_key("required")), has_entry("required", [])))34    assert_that(items, has_entry("properties", has_key("fork")))35def test_openapi_compatible_schema():36    schema = to_openapi_json_schema(test_objects[0], UpdateMode.GEN, schema=None)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.