Coverage for postrfp/web/suxint/openapi.py: 100%

1import textwrap 

2import importlib 

3from collections import defaultdict 

4from typing import Sequence 

5 

6from apispec import APISpec 

7from pydantic import BaseModel 

8from pydantic.json_schema import models_json_schema 

9 

10from postrfp.web.ext.apispec import SuxPlugin 

11from postrfp.shared.types import SuxType 

12 

13 

14def any_of_fix(prop: dict) -> dict: 

15 """ 

16 Convert this: 

17 { 

18 'anyOf: [ 

19 {'format': 'date-time', 'type': 'string'}, 

20 {'type': 'null'} 

21 ] 

22 } 

23 To: 

24 {"type": "integer", "format": 'date-time', 'nullable'; True "} 

25 """ 

26 new_dict = dict() 

27 

28 for any_of_option in prop["anyOf"]: 

29 if any_of_option.get("type") == "null": 

30 new_dict["nullable"] = True 

31 else: 

32 # Promote the main type specification from the anyOf list to be 

33 # the full specification 

34 new_dict.update(any_of_option) 

35 return new_dict 

36 

37 

38def servers_object(path: str) -> list[dict[str, str]]: 

39 return [ 

40 {"url": f"https://re.postrfp.com/{path}"}, 

41 {"url": f"http://localhost:9000/{path}"}, 

42 ] 

43 

44 

45def add_model_components(spec: APISpec, sux_instance: SuxType) -> None: 

46 if isinstance(sux_instance.models_module, str): 

47 models_mod = importlib.import_module(sux_instance.models_module) 

48 else: 

49 raise ValueError( 

50 "sux_instance models_module attribute must be a string module name" 

51 ) 

52 

53 mods: Sequence = [ 

54 (mod, "validation") 

55 for mod in (getattr(models_mod, m) for m in dir(models_mod)) 

56 if ( 

57 isinstance(mod, type) 

58 and issubclass(mod, BaseModel) 

59 and mod is not BaseModel 

60 ) 

61 ] 

62 

63 try: 

64 # This call to update_forward_refs is required because Nodes has recursive properties 

65 # - horrible hack because I cannot figure out how to determine if a model has been 

66 # created with call to typing.ForwardRef. See pydantic docs 

67 # https://pydantic-docs.helpmanual.io/usage/postponed_annotations/ 

68 getattr(models_mod, "Nodes").model_rebuild() 

69 except AttributeError: 

70 pass 

71 

72 _, fulldefs = models_json_schema(mods, ref_template="#/components/schemas/{model}") 

73 

74 defs = fulldefs["$defs"] 

75 for model_name in defs.keys(): 

76 model_spec = defs[model_name] 

77 spec.components.schema(model_name, model_spec) 

78 

79 

80# takes a Sux instance and returns a dictionary representation of the OpenAPI spec 

81def create_spec(sux_instance: SuxType, path: str = "api/") -> APISpec: 

82 from ..base import API_VERSION 

83 

84 spec = APISpec( 

85 title=sux_instance.api_name, 

86 info=dict(description=sux_instance.description), 

87 version=str(API_VERSION), 

88 openapi_version="3.1.1", 

89 plugins=[SuxPlugin()], 

90 servers=servers_object(path), 

91 ) 

92 

93 path_map = defaultdict(list) 

94 

95 extra_components = {} 

96 for handler in sux_instance.iter_handlers(): 

97 for adaptor in handler.adaptors(): 

98 if hasattr(adaptor, "enum_values") and adaptor.enum_values is not None: 

99 type_name = f"_{handler.name}{adaptor.arg_name.capitalize()}" 

100 extra_components[type_name] = { 

101 "title": type_name, 

102 "enum": list(adaptor.enum_values), 

103 "type": "string", 

104 } 

105 

106 path_map[handler.path].append(handler) 

107 

108 for url_key in sorted(path_map.keys()): 

109 handlers = path_map[url_key] 

110 sux_item = (url_key, handlers) 

111 spec.path(path=url_key, sux_handler=sux_item) 

112 

113 add_model_components(spec, sux_instance) 

114 # Add extra components for enum values 

115 for model_name in extra_components: 

116 spec.components.schema(model_name, extra_components[model_name]) 

117 

118 for mod_name, mod in sux_instance.modules.items(): 

119 tag_dict = {"name": mod_name.capitalize()} 

120 if mod.__doc__: 

121 tag_dict["description"] = textwrap.dedent(mod.__doc__).strip() 

122 

123 spec.tag(tag_dict) 

124 

125 return spec