Coverage for fastblocks / adapters / routes / default.py: 4%

1"""Default Routes Adapter for FastBlocks. 

2 

3Provides dynamic route discovery and registration for FastBlocks applications. 

4Includes automatic route gathering from adapters, static file serving, and HTMX endpoint support. 

5 

6Features: 

7- Dynamic route discovery from adapter modules 

8- HTMX-aware endpoints with template fragment rendering 

9- Built-in static routes (favicon, robots.txt) 

10- Automatic static file serving for storage adapters 

11- Template block rendering endpoints 

12- Route gathering from base routes.py files 

13- Integration with FastBlocks template system 

14 

15Requirements: 

16- starlette>=0.47.1 

17- jinja2>=3.1.6 

18 

19Usage: 

20```python 

21import typing as t 

22 

23from acb.depends import Inject, depends 

24from acb.adapters import import_adapter 

25 

26routes = depends.get("routes") 

27 

28Routes = import_adapter("routes") 

29 

30app_routes = routes.routes 

31``` 

32 

33Author: lesleslie <les@wedgwoodwebworks.com> 

34Created: 2025-01-12 

35""" 

36 

37import typing as t 

38from contextlib import suppress 

39from importlib import import_module 

40from uuid import UUID 

41 

42try: 

43 from acb.adapters import ( 

44 AdapterStatus, 

45 get_adapters, 

46 get_installed_adapter, 

47 import_adapter, 

48 root_path, 

49 ) 

50except ImportError: # acb >= 0.19 removed get_installed_adapter 

51 from acb.adapters import ( 

52 AdapterStatus, 

53 get_adapter, 

54 get_adapters, 

55 get_installed_adapters, 

56 import_adapter, 

57 root_path, 

58 ) 

59 

60 def get_installed_adapter(adapter_name: str) -> str | None: 

61 """Compatibility shim that resolves an installed adapter name.""" 

62 for adapter in get_installed_adapters(): 

63 meta = getattr(adapter, "metadata", None) 

64 provider = getattr(meta, "provider", None) 

65 if adapter_name in (adapter.category, adapter.name, provider): 

66 return provider or adapter.name 

67 adapter = get_adapter(adapter_name) 

68 if adapter: 

69 meta = getattr(adapter, "metadata", None) 

70 provider = getattr(meta, "provider", None) 

71 return provider or adapter.name 

72 return None 

73 

74 

75from acb.config import Config 

76from acb.debug import debug 

77from acb.depends import depends 

78from anyio import Path as AsyncPath 

79from jinja2.exceptions import TemplateNotFound 

80from starlette.endpoints import HTTPEndpoint 

81from starlette.exceptions import HTTPException 

82from starlette.requests import Request 

83from starlette.responses import PlainTextResponse, Response 

84from starlette.routing import Host, Mount, Route, Router, WebSocketRoute 

85from starlette.types import Receive, Scope, Send 

86from fastblocks.actions.query import create_query_context 

87from fastblocks.htmx import HtmxRequest 

88 

89from ._base import RoutesBase, RoutesBaseSettings 

90 

91try: 

92 Templates = import_adapter("templates") 

93except Exception: 

94 Templates = None 

95 

96base_routes_path = root_path / "routes.py" 

97 

98 

99class RoutesSettings(RoutesBaseSettings): ... 

100 

101 

102class FastBlocksEndpoint(HTTPEndpoint): 

103 def __init__( 

104 self, 

105 scope: Scope, 

106 receive: Receive, 

107 send: Send, 

108 config: Config | None = None, 

109 ) -> None: 

110 super().__init__(scope, receive, send) 

111 self.config = config or depends.get_sync(Config) 

112 self.templates = depends.get_sync("templates") 

113 

114 

115class Index(FastBlocksEndpoint): 

116 async def get(self, request: HtmxRequest | Request) -> Response: 

117 debug(request) 

118 path_params = getattr(request, "path_params", {}) 

119 page = path_params.get("page") or "home" 

120 template = "index.html" 

121 headers = {"vary": "hx-request"} 

122 scope = getattr(request, "scope", {}) 

123 if htmx := scope.get("htmx"): 

124 debug(htmx) 

125 template = f"{page.lstrip('/')}.html" 

126 headers["hx-push-url"] = "/" if page == "home" else page 

127 debug(page, template) 

128 context = await create_query_context( 

129 request, base_context={"page": page.lstrip("/")} 

130 ) 

131 query_params = getattr(request, "query_params", {}) 

132 if "model" in query_params: 

133 model_name = query_params["model"] 

134 if f"{model_name}_parser" in context: 

135 parser = context[f"{model_name}_parser"] 

136 context[f"{model_name}_list"] = await parser.parse_and_execute() 

137 context[f"{model_name}_count"] = await parser.get_count() 

138 try: 

139 result = await self.templates.render_template( 

140 request, 

141 template, 

142 headers=headers, 

143 context=context, 

144 ) 

145 return t.cast(Response, result) 

146 except TemplateNotFound: 

147 raise HTTPException(status_code=404) 

148 

149 

150class Block(FastBlocksEndpoint): 

151 async def get(self, request: HtmxRequest | Request) -> Response: 

152 debug(request) 

153 path_params = getattr(request, "path_params", {}) 

154 block = f"blocks/{path_params.get('block', 'default')}.html" 

155 context = await create_query_context(request) 

156 query_params = getattr(request, "query_params", {}) 

157 if "model" in query_params: 

158 model_name = query_params["model"] 

159 if f"{model_name}_parser" in context: 

160 parser = context[f"{model_name}_parser"] 

161 context[f"{model_name}_list"] = await parser.parse_and_execute() 

162 context[f"{model_name}_count"] = await parser.get_count() 

163 try: 

164 result = await self.templates.render_template( 

165 request, block, context=context 

166 ) 

167 return t.cast(Response, result) 

168 except TemplateNotFound: 

169 raise HTTPException(status_code=404) 

170 

171 

172class Component(FastBlocksEndpoint): 

173 async def get(self, request: HtmxRequest | Request) -> Response: 

174 debug(request) 

175 component_name = getattr(request, "path_params", {}).get("component", "default") 

176 query_params = getattr(request, "query_params", {}) 

177 context = await create_query_context(request, base_context=dict(query_params)) 

178 if "model" in query_params: 

179 model_name = query_params["model"] 

180 if f"{model_name}_parser" in context: 

181 parser = context[f"{model_name}_parser"] 

182 context[f"{model_name}_list"] = await parser.parse_and_execute() 

183 context[f"{model_name}_count"] = await parser.get_count() 

184 try: 

185 htmy = await depends.get("htmy") 

186 if htmy is None: 

187 raise HTTPException( 

188 status_code=500, detail="HTMY adapter not available" 

189 ) 

190 result = await htmy.render_component( 

191 request, component_name, context=context 

192 ) 

193 return t.cast(Response, result) 

194 except Exception as e: 

195 debug(f"Component '{component_name}' not found: {e}") 

196 raise HTTPException(status_code=404) 

197 

198 

199class Routes(RoutesBase): 

200 routes: list[Route | Router | Mount | Host | WebSocketRoute] = [] 

201 

202 async def gather_routes(self, path: AsyncPath) -> None: 

203 depth = -2 

204 if "adapters" in path.parts: 

205 depth = -4 

206 module_path = ".".join(path.parts[depth:]).removesuffix(".py") 

207 debug(path, depth, module_path) 

208 with suppress(ModuleNotFoundError): 

209 module = import_module(module_path) 

210 module_routes = getattr(module, "routes", None) 

211 if module_routes and isinstance(module_routes, list): 

212 self.routes = module.routes + self.routes 

213 

214 @staticmethod 

215 async def favicon(request: Request) -> Response: 

216 return PlainTextResponse("", 200) 

217 

218 @staticmethod 

219 async def robots(request: Request) -> Response: 

220 txt = "User-agent: *\nDisallow: /dashboard/\nDisallow: /blocks/" 

221 return PlainTextResponse(txt, 200) 

222 

223 async def init(self) -> None: 

224 self.routes.extend( 

225 [ 

226 Route("/favicon.ico", endpoint=self.favicon, methods=["GET"]), 

227 Route("/robots.txt", endpoint=self.robots, methods=["GET"]), 

228 Route("/", Index, methods=["GET"]), 

229 Route("/{page}", Index, methods=["GET"]), 

230 Route("/block/{block}", Block, methods=["GET"]), 

231 Route("/component/{component}", Component, methods=["GET"]), 

232 ], 

233 ) 

234 for adapter in get_adapters(): 

235 routes_path = adapter.path.parent / "_routes.py" 

236 if await routes_path.exists(): 

237 await self.gather_routes(routes_path) 

238 if await base_routes_path.exists(): 

239 await self.gather_routes(base_routes_path) 

240 if get_installed_adapter("storage") in ("file", "memory"): 

241 from starlette.staticfiles import StaticFiles 

242 

243 self.routes.append( 

244 Mount( 

245 "/media", 

246 app=StaticFiles(directory=self.config.storage.local_path / "media"), 

247 name="media", 

248 ), 

249 ) 

250 if not self.config.deployed: 

251 from starlette.staticfiles import StaticFiles 

252 

253 self.routes.append( 

254 Mount( 

255 "/static", 

256 app=StaticFiles( 

257 directory=self.config.storage.local_path / "static" 

258 ), 

259 name="media", 

260 ), 

261 ) 

262 debug(self.routes) 

263 

264 

265MODULE_ID = UUID("01937d86-6f4c-7d5e-a01f-3456789012cd") 

266MODULE_STATUS = AdapterStatus.STABLE 

267 

268with suppress(Exception): 

269 depends.set(Routes, "default")