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 fastapi import Body, FastAPI, HTTPException 

10from pydantic import BaseModel, ConfigDict, ValidationError 

11 

12 

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

14 app: FastAPI, 

15 path_prefix: str, 

16 *, 

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

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

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

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

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

22) -> None: 

23 path_prefix = path_prefix.rstrip("/") 

24 

25 if create_func: 

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

27 async def create_endpoint( 

28 identifier: KT, 

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

30 ) -> VT | None: 

31 return await create_func( 

32 identifier, 

33 put_body, 

34 ) 

35 

36 if read_func: 

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

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

39 return await read_func( 

40 identifier, 

41 ) 

42 

43 if update_func: 

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

45 async def update_endpoint( 

46 identifier: KT, 

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

48 ) -> VT | None: 

49 return await update_func( 

50 identifier, 

51 patch_body, 

52 ) 

53 

54 if delete_func: 

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

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

57 return await delete_func(identifier) 

58 

59 if list_func: 

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

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

62 try: 

63 return dict(await list_func( 

64 json.loads(filter_body) if filter_body 

65 else None, 

66 )) 

67 except json.JSONDecodeError as e: 

68 raise HTTPException(HTTPStatus.BAD_REQUEST) from e 

69 

70 

71class FilterKeys(BaseModel): 

72 model_config = ConfigDict(extra="forbid") 

73 

74 includes: list[str] | None = None 

75 excludes: list[str] | None = None 

76 

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

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

79 

80 

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

82 app: FastAPI, 

83 path_prefix: str, 

84 mapping: Mapping[KT, VT], 

85 *, 

86 values_in_list: bool = False, 

87 readonly: bool | None = None, 

88) -> None: 

89 mutable: bool = isinstance(mapping, MutableMapping) 

90 if readonly is None: 

91 readonly = not mutable 

92 

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

94 try: 

95 return crudl_store.read(key) 

96 except KeyError as e: 

97 raise HTTPException(HTTPStatus.NOT_FOUND) from e 

98 

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

100 try: 

101 filter_keys_model: FilterKeys | None = ( 

102 FilterKeys.model_validate(filter_keys, strict=True) if filter_keys 

103 else None 

104 ) 

105 except ValidationError as e: 

106 raise HTTPException(HTTPStatus.BAD_REQUEST) from e 

107 

108 return crudl_store.list( 

109 None if filter_keys_model is None 

110 else filter_keys_model.match, 

111 ) 

112 

113 if mutable and not readonly: 

114 crudl_store = CRUDLWrapper[KT, VT]( 

115 mapping, 

116 values_in_list=values_in_list, 

117 ) 

118 

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

120 try: 

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

122 except KeyError as e: 

123 raise HTTPException(HTTPStatus.CONFLICT) from e 

124 

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

126 try: 

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

128 except KeyError as e: 

129 raise HTTPException(HTTPStatus.NOT_FOUND) from e 

130 

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

132 try: 

133 crudl_store.delete(key) 

134 except KeyError as e: 

135 raise HTTPException(HTTPStatus.NOT_FOUND) from e 

136 

137 add_crudl_endpoints( 

138 app, 

139 path_prefix, 

140 read_func=read_func, 

141 list_func=list_func, 

142 create_func=create_func, 

143 update_func=update_func, 

144 delete_func=delete_func, 

145 ) 

146 else: 

147 crudl_store = RLWrapper[KT, VT]( 

148 mapping, 

149 values_in_list=values_in_list, 

150 ) 

151 

152 add_crudl_endpoints( 

153 app, 

154 path_prefix, 

155 read_func=read_func, 

156 list_func=list_func, 

157 )