Coverage for src/extratools_api/crudl.py: 0%

1import json 

2from collections.abc import Awaitable, Callable, Iterable, Mapping, MutableMapping 

3from http import HTTPStatus 

4from typing import Annotated, Any, cast 

5 

6from extratools_core.crudl import CRUDLWrapper, RLWrapper 

7from extratools_core.json import JsonDict 

8from extratools_core.str import wildcard_match 

9from extratools_core.typing import SearchableMapping 

10from fastapi import Body, FastAPI, HTTPException 

11from pydantic import BaseModel, ConfigDict, ValidationError 

12 

13 

14def add_crudl_endpoints[KT: str, VT: JsonDict | BaseModel]( 

15 app: FastAPI, 

16 path_prefix: str, 

17 *, 

18 create_func: Callable[[KT, JsonDict], Awaitable[VT | None]] | None = None, 

19 read_func: Callable[[KT], Awaitable[VT]] | None = None, 

20 update_func: Callable[[KT, JsonDict], Awaitable[VT | None]] | None = None, 

21 delete_func: Callable[[KT], Awaitable[VT | None]] | None = None, 

22 list_func: Callable[[JsonDict | None], Awaitable[Iterable[tuple[KT, Any]]]] | None = None, 

23) -> None: 

24 path_prefix = path_prefix.rstrip("/") 

25 

26 if create_func: 

27 @app.put(path_prefix + "/{identifier}") 

28 async def create_endpoint( 

29 identifier: KT, 

30 put_body: Annotated[JsonDict, Body()], 

31 ) -> VT | None: 

32 return await create_func( 

33 identifier, 

34 put_body, 

35 ) 

36 

37 if read_func: 

38 @app.get(path_prefix + "/{identifier}") 

39 async def read_endpoint(identifier: KT) -> VT: 

40 return await read_func( 

41 identifier, 

42 ) 

43 

44 if update_func: 

45 @app.patch(path_prefix + "/{identifier}") 

46 async def update_endpoint( 

47 identifier: KT, 

48 patch_body: Annotated[JsonDict, Body()], 

49 ) -> VT | None: 

50 return await update_func( 

51 identifier, 

52 patch_body, 

53 ) 

54 

55 if delete_func: 

56 @app.delete(path_prefix + "/{identifier}") 

57 async def delete_endpoint(identifier: KT) -> VT | None: 

58 return await delete_func(identifier) 

59 

60 if list_func: 

61 @app.get(path_prefix + "/") 

62 async def list_endpoint(filter_body: str | None = None) -> dict[KT, Any]: 

63 try: 

64 return dict(await list_func( 

65 json.loads(filter_body) if filter_body 

66 else None, 

67 )) 

68 except json.JSONDecodeError as e: 

69 raise HTTPException(HTTPStatus.BAD_REQUEST) from e 

70 

71 

72class FilterBody(BaseModel): 

73 model_config = ConfigDict(extra="forbid") 

74 

75 query: Any = None 

76 

77 includes: list[str] | None = None 

78 excludes: list[str] | None = None 

79 

80 def match(self, key: str) -> bool: 

81 return wildcard_match(key, includes=self.includes, excludes=self.excludes) 

82 

83 

84def add_crudl_endpoints_for_mapping[KT: str, VT: JsonDict]( 

85 app: FastAPI, 

86 path_prefix: str, 

87 mapping: Mapping[KT, VT], 

88 *, 

89 values_in_list: bool = False, 

90 readonly: bool | None = None, 

91) -> None: 

92 mutable: bool = isinstance(mapping, MutableMapping) 

93 if readonly is None: 

94 readonly = not mutable 

95 

96 async def read_func(key: KT) -> VT: 

97 try: 

98 return crudl_store.read(key) 

99 except KeyError as e: 

100 raise HTTPException(HTTPStatus.NOT_FOUND) from e 

101 

102 async def list_func(filter_body: JsonDict | None) -> Iterable[tuple[KT, VT | None]]: 

103 try: 

104 filter_body_model: FilterBody | None = ( 

105 FilterBody.model_validate(filter_body, strict=True) if filter_body 

106 else None 

107 ) 

108 except ValidationError as e: 

109 raise HTTPException(HTTPStatus.BAD_REQUEST) from e 

110 

111 if ( 

112 filter_body_model 

113 and filter_body_model.query is not None 

114 and not isinstance(mapping, SearchableMapping) 

115 ): 

116 raise HTTPException(HTTPStatus.BAD_REQUEST) 

117 

118 return crudl_store.list( 

119 None if filter_body_model is None 

120 else ( 

121 filter_body_model.query if isinstance(mapping, SearchableMapping) else None, 

122 filter_body_model.match, 

123 ), 

124 ) 

125 

126 if mutable and not readonly: 

127 crudl_store = CRUDLWrapper[KT, VT]( 

128 mapping, 

129 values_in_list=values_in_list, 

130 ) 

131 

132 async def create_func(key: KT, value: JsonDict) -> None: 

133 try: 

134 crudl_store.create(key, cast("VT", value)) 

135 except KeyError as e: 

136 raise HTTPException(HTTPStatus.CONFLICT) from e 

137 

138 async def update_func(key: KT, value: JsonDict) -> None: 

139 try: 

140 crudl_store.update(key, cast("VT", value)) 

141 except KeyError as e: 

142 raise HTTPException(HTTPStatus.NOT_FOUND) from e 

143 

144 async def delete_func(key: KT) -> None: 

145 try: 

146 crudl_store.delete(key) 

147 except KeyError as e: 

148 raise HTTPException(HTTPStatus.NOT_FOUND) from e 

149 

150 add_crudl_endpoints( 

151 app, 

152 path_prefix, 

153 read_func=read_func, 

154 list_func=list_func, 

155 create_func=create_func, 

156 update_func=update_func, 

157 delete_func=delete_func, 

158 ) 

159 else: 

160 crudl_store = RLWrapper[KT, VT]( 

161 mapping, 

162 values_in_list=values_in_list, 

163 ) 

164 

165 add_crudl_endpoints( 

166 app, 

167 path_prefix, 

168 read_func=read_func, 

169 list_func=list_func, 

170 )