Coverage for postrfp/web/ext/apispec.py: 92%

1""" 

2Integration with apispec 

3 

4http://apispec.readthedocs.org/en/latest/ 

5 

6This module implements an apispec plugin which introspects a suxint.Sux 

7instance enabling apispec to produce OpenAPI compatible json documentation 

8 

9https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.3.md 

10 

11""" 

12 

13from inspect import Signature 

14from typing import TYPE_CHECKING, Any 

15 

16if TYPE_CHECKING: 

17 from postrfp.web.suxint import Handler 

18from postrfp.shared.constants import MimeTypes 

19from apispec import BasePlugin 

20 

21from pydantic import BaseModel 

22 

23from .apilinks import Link 

24from .openapi_types import openapi_type_mapping 

25 

26 

27def responses(handler: "Handler"): 

28 from postrfp.shared.response import XAccelResponse 

29 

30 rdict: dict[str, dict[str, Any]] = {"200": {"description": "Successful Response"}} 

31 

32 api = ( 

33 "buyer" if handler.func.__module__.startswith("postrfp.buyer.api") else "vendor" 

34 ) 

35 link_key = (api, handler.name) 

36 if link_key in Link.instances: 

37 rdict["200"]["links"] = Link.instances.get(link_key, []) 

38 

39 retval = handler.return_annotation or handler.return_value 

40 

41 if retval is Signature.empty or retval is None: 

42 return rdict 

43 

44 if isinstance(retval, type) and issubclass(retval, BaseModel): 

45 model_name = retval.__name__ 

46 schema: dict[str, str | dict] = {"$ref": f"#/components/schemas/{model_name}"} 

47 

48 elif getattr(retval, "__origin__", None) is list: 

49 # __origin__ is set by the typing module with List[x] 

50 item_type_name = retval.__args__[0].__name__ 

51 if item_type_name in openapi_type_mapping: 

52 items_dict = {"type": openapi_type_mapping[item_type_name].value} 

53 else: 

54 items_dict = {"$ref": f"#/components/schemas/{retval.__args__[0].__name__}"} 

55 schema = {"type": "array", "items": items_dict} 

56 elif retval is XAccelResponse: 

57 rdict["200"]["content"] = { 

58 MimeTypes.ANY.value: {"schema": {"type": "string", "format": "binary"}} 

59 } 

60 return rdict 

61 else: 

62 msg = "Error generating API spec - API function annotations must be" 

63 msg += " Pydantic schema definitions or List[Pydantic Model]." 

64 msg += f" The return value '{str(retval)}' of {handler} is not recognised." 

65 raise TypeError(msg) 

66 

67 rdict["200"]["content"] = {"application/json": {"schema": schema}} 

68 

69 return rdict 

70 

71 

72def operations_from_handler(handler): 

73 path_object = { 

74 "description": handler.docs, 

75 "parameters": [], 

76 "responses": responses(handler), 

77 "operationId": handler.name, 

78 } 

79 if handler.tag: 

80 path_object["tags"] = [handler.tag] 

81 for adaptor in handler.adaptors(): 

82 if hasattr(adaptor, "update_openapi_path_object"): 

83 adaptor.update_openapi_path_object(path_object) 

84 

85 return path_object 

86 

87 

88class SuxPlugin(BasePlugin): 

89 def init_spec(self, spec): 

90 super(SuxPlugin, self).init_spec(spec) 

91 self.openapi_major_version = spec.openapi_version.major 

92 

93 def path_helper(self, path, sux_handler, operations=None, **kwargs): 

94 """Path helper that adapts a Sux handler to a Path""" 

95 _sux_path, sux_handlers = sux_handler 

96 for handler in sux_handlers: 

97 operations[handler.method.lower()] = operations_from_handler(handler)