Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/flask_restx/resource.py: 27%

1from flask import request 

2from flask.views import MethodView 

3from werkzeug import __version__ as werkzeug_version 

4 

5if werkzeug_version.split(".")[0] >= "2": 

6 from werkzeug.wrappers import Response as BaseResponse 

7else: 

8 from werkzeug.wrappers import BaseResponse 

9 

10from .model import ModelBase 

11 

12from .utils import unpack 

13 

14 

15class Resource(MethodView): 

16 """ 

17 Represents an abstract RESTX resource. 

18 

19 Concrete resources should extend from this class 

20 and expose methods for each supported HTTP method. 

21 If a resource is invoked with an unsupported HTTP method, 

22 the API will return a response with status 405 Method Not Allowed. 

23 Otherwise the appropriate method is called and passed all arguments 

24 from the url rule used when adding the resource to an Api instance. 

25 See :meth:`~flask_restx.Api.add_resource` for details. 

26 """ 

27 

28 representations = None 

29 method_decorators = [] 

30 

31 def __init__(self, api=None, *args, **kwargs): 

32 self.api = api 

33 

34 def dispatch_request(self, *args, **kwargs): 

35 # Taken from flask 

36 meth = getattr(self, request.method.lower(), None) 

37 if meth is None and request.method == "HEAD": 

38 meth = getattr(self, "get", None) 

39 assert meth is not None, "Unimplemented method %r" % request.method 

40 

41 for decorator in self.method_decorators: 

42 meth = decorator(meth) 

43 

44 self.validate_payload(meth) 

45 

46 resp = meth(*args, **kwargs) 

47 

48 if isinstance(resp, BaseResponse): 

49 return resp 

50 

51 representations = self.representations or {} 

52 

53 mediatype = request.accept_mimetypes.best_match(representations, default=None) 

54 if mediatype in representations: 

55 data, code, headers = unpack(resp) 

56 resp = representations[mediatype](data, code, headers) 

57 resp.headers["Content-Type"] = mediatype 

58 return resp 

59 

60 return resp 

61 

62 def __validate_payload(self, expect, collection=False): 

63 """ 

64 :param ModelBase expect: the expected model for the input payload 

65 :param bool collection: False if a single object of a resource is 

66 expected, True if a collection of objects of a resource is expected. 

67 """ 

68 # TODO: proper content negotiation 

69 data = request.get_json() 

70 if collection: 

71 data = data if isinstance(data, list) else [data] 

72 for obj in data: 

73 expect.validate(obj, self.api.refresolver, self.api.format_checker) 

74 else: 

75 expect.validate(data, self.api.refresolver, self.api.format_checker) 

76 

77 def validate_payload(self, func): 

78 """Perform a payload validation on expected model if necessary""" 

79 if getattr(func, "__apidoc__", False) is not False: 

80 doc = func.__apidoc__ 

81 validate = doc.get("validate", None) 

82 validate = validate if validate is not None else self.api._validate 

83 if validate: 

84 for expect in doc.get("expect", []): 

85 # TODO: handle third party handlers 

86 if isinstance(expect, list) and len(expect) == 1: 

87 if isinstance(expect[0], ModelBase): 

88 self.__validate_payload(expect[0], collection=True) 

89 if isinstance(expect, ModelBase): 

90 self.__validate_payload(expect, collection=False)