Coverage for rfpy/web/serial/extractor.py: 100%

1# pylint: disable=E0213 

2import logging 

3 

4from pydantic import ValidationError 

5 

6from rfpy.suxint import ArgExtractor 

7from rfpy.model.exc import ValidationFailure 

8 

9log = logging.getLogger(__name__) 

10 

11""" 

12 Pydantic ALIAS 

13 

14 - alias should be the public name if different from the db column name 

15 - when dumping/serialising when outputting data from the data to the outside world 

16 use model.dict(by_alias) 

17 - when parsing incoming data, use model.model_dump() 

18 

19 by_alias == False (the default) means 'dump data for consumption by database' 

20 by_alias == True means 'dump data for use by API consumers' (default behaviour) 

21 

22""" 

23 

24 

25class SchemaDocArg(ArgExtractor): 

26 swagger_in = "body" 

27 

28 def __init__(self, schema_cls, as_dict=True, exclude_unset=False): 

29 self.as_dict = as_dict 

30 self.exclude_unset = exclude_unset 

31 super().__init__(None) 

32 self.schema_cls = schema_cls 

33 

34 def extract(self, request): 

35 try: 

36 # NB exclude_unset=True is important. If not set, or False, 

37 # then all the pydantic Model attributes not set will 

38 # appear in the data passed to API methods - i.e. lots of None 

39 # values which can overwrite real data on update 

40 pydantic_model = self.schema_cls.model_validate(request.json) 

41 if not self.as_dict: 

42 return pydantic_model 

43 else: 

44 return pydantic_model.model_dump(exclude_unset=self.exclude_unset) 

45 

46 except ValidationError as e: 

47 log.debug("Schema validation failed: %s", e) 

48 err_list = [] 

49 for err in e.errors(): 

50 location = " > ".join(f"'{e}'" for e in err["loc"]) 

51 msg = err["msg"].title() 

52 err_msg = f"Validation failed. {msg}: {location}" 

53 err_list.append(err_msg) 

54 raise ValidationFailure(str(e), err_list) 

55 

56 def update_openapi_path_object(self, path_obj): 

57 path_obj["requestBody"] = { 

58 "content": { 

59 "application/json": { 

60 "schema": { 

61 "$ref": "#/components/schemas/%s" % self.schema_cls.__name__ 

62 } 

63 } 

64 } 

65 }