Coverage for fastblocks / adapters / templates / _base.py: 65%

1import typing as t 

2from abc import ABC 

3 

4from acb.adapters import get_adapters, root_path 

5from acb.config import AdapterBase, Config 

6from acb.depends import Inject, depends 

7from anyio import Path as AsyncPath 

8from starlette.requests import Request 

9from starlette.responses import Response 

10 

11try: 

12 # For newer versions of ACB where pkg_registry is in context 

13 from acb import get_context 

14 

15 context = get_context() 

16 pkg_registry = context.pkg_registry 

17except (ImportError, AttributeError): 

18 # Fallback for older versions or if context is not available 

19 pkg_registry = None 

20 

21 

22async def safe_await(func_or_value: t.Any) -> t.Any: 

23 if callable(func_or_value): 

24 try: 

25 result = func_or_value() 

26 if hasattr(result, "__await__") and callable(result.__await__): 

27 return await t.cast("t.Awaitable[t.Any]", result) 

28 return result 

29 except Exception: 

30 return True 

31 return func_or_value 

32 

33 

34TemplateContext: t.TypeAlias = dict[str, t.Any] 

35TemplateResponse: t.TypeAlias = Response 

36TemplateStr: t.TypeAlias = str 

37TemplatePath: t.TypeAlias = str 

38T = t.TypeVar("T") 

39 

40 

41class TemplateRenderer(t.Protocol): 

42 async def render_template( 

43 self, 

44 request: Request, 

45 template: TemplatePath, 

46 _: TemplateContext | None = None, 

47 ) -> TemplateResponse: ... 

48 

49 

50class TemplateLoader(t.Protocol): 

51 async def get_template(self, name: TemplatePath) -> t.Any: ... 

52 

53 async def list_templates(self) -> list[TemplatePath]: ... 

54 

55 

56class TemplatesBaseSettings(Config, ABC): 

57 cache_timeout: int = 300 

58 

59 @depends.inject 

60 def __init__(self, config: Inject[Config], **values: t.Any) -> None: 

61 # Extract cache_timeout from values before passing to parent 

62 cache_timeout = values.pop("cache_timeout", 300) 

63 super().__init__(**values) 

64 self.cache_timeout = cache_timeout if config.deployed else 1 

65 

66 

67class TemplatesProtocol(t.Protocol): 

68 def get_searchpath(self, adapter: t.Any, path: AsyncPath) -> None: ... 

69 

70 async def get_searchpaths(self, adapter: t.Any) -> list[AsyncPath]: ... 

71 

72 @staticmethod 

73 def get_storage_path(path: AsyncPath) -> AsyncPath: ... 

74 

75 @staticmethod 

76 def get_cache_key(path: AsyncPath) -> str: ... 

77 

78 

79class TemplatesBase(AdapterBase): 

80 app: t.Any | None = None 

81 admin: t.Any | None = None 

82 app_searchpaths: list[AsyncPath] | None = None 

83 admin_searchpaths: list[AsyncPath] | None = None 

84 

85 def get_searchpath(self, adapter: t.Any, path: AsyncPath) -> list[AsyncPath]: 

86 style = getattr(self.config.app, "style", "bulma") 

87 base_path = path / "base" 

88 style_path = path / style 

89 style_adapter_path = path / style / adapter.name 

90 theme_adapter_path = style_adapter_path / "theme" 

91 return [theme_adapter_path, style_adapter_path, style_path, base_path] 

92 

93 async def get_searchpaths(self, adapter: t.Any) -> list[AsyncPath]: 

94 searchpaths = [] 

95 base_root = self._get_base_root() 

96 

97 if adapter and hasattr(adapter, "category"): 

98 searchpaths.extend( 

99 self.get_searchpath( 

100 adapter, base_root / "templates" / adapter.category 

101 ), 

102 ) 

103 

104 if adapter and hasattr(adapter, "category") and adapter.category == "app": 

105 searchpaths.extend(await self._get_app_searchpaths(adapter)) 

106 

107 # Only use pkg_registry if it's available 

108 if pkg_registry: 

109 searchpaths.extend(await self._get_pkg_registry_searchpaths(adapter)) 

110 

111 return searchpaths 

112 

113 def _get_base_root(self) -> AsyncPath: 

114 if callable(root_path): 

115 return AsyncPath(root_path()) 

116 return AsyncPath(root_path) 

117 

118 async def _get_app_searchpaths(self, adapter: t.Any) -> list[AsyncPath]: 

119 searchpaths = [] 

120 for a in ( 

121 a 

122 for a in get_adapters() 

123 if a 

124 and hasattr(a, "category") 

125 and a.category not in ("app", "admin", "secret") 

126 ): 

127 exists_result = await safe_await((a.path / "_templates").exists) 

128 if exists_result: 

129 searchpaths.append(a.path / "_templates") 

130 return searchpaths 

131 

132 async def _get_pkg_registry_searchpaths(self, adapter: t.Any) -> list[AsyncPath]: 

133 searchpaths = [] 

134 for pkg in pkg_registry.get(): 

135 if ( 

136 pkg 

137 and hasattr(pkg, "path") 

138 and adapter 

139 and hasattr(adapter, "category") 

140 ): 

141 searchpaths.extend( 

142 self.get_searchpath( 

143 adapter, 

144 pkg.path / "adapters" / adapter.category / "_templates", 

145 ), 

146 ) 

147 return searchpaths 

148 

149 @staticmethod 

150 def get_storage_path(path: AsyncPath) -> AsyncPath: 

151 templates_path_name = "templates" 

152 if templates_path_name not in path.parts: 

153 templates_path_name = "_templates" 

154 depth = path.parts.index(templates_path_name) - 1 

155 _path = list(path.parts[depth:]) 

156 _path.insert(1, _path.pop(0)) 

157 return AsyncPath("/".join(_path)) 

158 depth = path.parts.index(templates_path_name) 

159 return AsyncPath("/".join(path.parts[depth:])) 

160 

161 @staticmethod 

162 def get_cache_key(path: AsyncPath) -> str: 

163 return ":".join(path.parts)