How to use to_json_schema method in pandera

Best Python code snippet using pandera_python

restapi.py

Source:restapi.py Github

copy

Full Screen

...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):...

Full Screen

Full Screen

typing.py

Source:typing.py Github

copy

Full Screen

...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()...

Full Screen

Full Screen

test_json_schema.py

Source:test_json_schema.py Github

copy

Full Screen

...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)...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run pandera automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful