Coverage for fastblocks / adapters / templates / hybrid.py: 28%
152 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-26 03:30 -0800
1"""Hybrid Template Management Integration Module.
3This module integrates all Hybrid template management components:
4- Hybrid Template Manager with validation and autocomplete
5- Async Template Renderer with performance optimization
6- Block Renderer for HTMX fragments and partials
7- Enhanced filters for secondary adapters
8- Template CLI tools and utilities
10This provides a unified interface for FastBlocks Week 7-8 template features.
12Requirements:
13- jinja2>=3.1.6
14- jinja2-async-environment>=0.14.3
15- starlette-async-jinja>=1.12.4
17Author: lesleslie <les@wedgwoodwebworks.com>
18Created: 2025-01-12
19"""
21import typing as t
22from contextlib import suppress
23from uuid import UUID
25from acb.adapters import AdapterStatus
26from acb.depends import depends
27from starlette.requests import Request
28from starlette.responses import Response
30from ._advanced_manager import HybridTemplatesManager, HybridTemplatesSettings
31from ._async_filters import FASTBLOCKS_ASYNC_FILTERS
32from ._async_renderer import AsyncTemplateRenderer, RenderContext, RenderMode
33from ._block_renderer import BlockRenderer, BlockRenderRequest, BlockUpdateMode
34from ._enhanced_filters import ENHANCED_ASYNC_FILTERS, ENHANCED_FILTERS
35from ._filters import FASTBLOCKS_FILTERS
36from .jinja2 import Templates
39class HybridTemplates:
40 """Unified interface for Hybrid template management features."""
42 def __init__(self) -> None:
43 self.settings = HybridTemplatesSettings()
44 self.base_templates: Templates | None = None
45 self.hybrid_manager: HybridTemplatesManager | None = None
46 self.async_renderer: AsyncTemplateRenderer | None = None
47 self.block_renderer: BlockRenderer | None = None
48 self._initialized = False
50 async def initialize(self) -> None:
51 """Initialize all components."""
52 if self._initialized:
53 return
55 # Get or create base templates
56 try:
57 self.base_templates = depends.get("templates")
58 except Exception:
59 self.base_templates = Templates()
60 await self.base_templates.init()
62 # Initialize Hybrid manager
63 self.hybrid_manager = HybridTemplatesManager(self.settings)
64 await self.hybrid_manager.initialize()
66 # Initialize async renderer
67 self.async_renderer = AsyncTemplateRenderer(
68 base_templates=self.base_templates, hybrid_manager=self.hybrid_manager
69 )
70 await self.async_renderer.initialize()
72 # Initialize block renderer
73 self.block_renderer = BlockRenderer(
74 async_renderer=self.async_renderer, hybrid_manager=self.hybrid_manager
75 )
76 await self.block_renderer.initialize()
78 # Register all filters
79 await self._register_filters()
81 self._initialized = True
83 async def _register_filters(self) -> None:
84 """Register all template filters with Jinja2 environments."""
85 if not self.base_templates:
86 return
88 all_filters = FASTBLOCKS_FILTERS | ENHANCED_FILTERS
90 all_async_filters = FASTBLOCKS_ASYNC_FILTERS | ENHANCED_ASYNC_FILTERS
92 # Register with main app environment
93 if self.base_templates.app and hasattr(self.base_templates.app.env, "filters"):
94 for name, filter_func in all_filters.items():
95 self.base_templates.app.env.filters[name] = filter_func
97 for name, async_filter_func in all_async_filters.items():
98 self.base_templates.app.env.filters[name] = async_filter_func
100 # Register with admin environment if available
101 if self.base_templates.admin and hasattr(
102 self.base_templates.admin.env, "filters"
103 ):
104 for name, filter_func in all_filters.items():
105 self.base_templates.admin.env.filters[name] = filter_func
107 for name, async_filter_func in all_async_filters.items():
108 self.base_templates.admin.env.filters[name] = async_filter_func
110 # Template Validation API
111 async def validate_template(
112 self,
113 template_source: str,
114 template_name: str = "unknown",
115 context: dict[str, t.Any] | None = None,
116 ) -> dict[str, t.Any]:
117 """Validate template and return results."""
118 if not self.hybrid_manager:
119 await self.initialize()
120 assert self.hybrid_manager is not None
122 result = await self.hybrid_manager.validate_template(
123 template_source, template_name, context
124 )
126 return {
127 "is_valid": result.is_valid,
128 "errors": [
129 {
130 "message": error.message,
131 "line_number": error.line_number,
132 "column_number": error.column_number,
133 "error_type": error.error_type,
134 "severity": error.severity,
135 "context": error.context,
136 }
137 for error in result.errors
138 ],
139 "warnings": [
140 {
141 "message": warning.message,
142 "line_number": warning.line_number,
143 "severity": warning.severity,
144 }
145 for warning in result.warnings
146 ],
147 "suggestions": result.suggestions,
148 "used_variables": list(result.used_variables),
149 "undefined_variables": list(result.undefined_variables),
150 "available_filters": list(result.available_filters),
151 "available_functions": list(result.available_functions),
152 }
154 # Autocomplete API
155 async def get_autocomplete_suggestions(
156 self, context: str, cursor_position: int = 0, template_name: str = "unknown"
157 ) -> list[dict[str, t.Any]]:
158 """Get autocomplete suggestions for template editing."""
159 if not self.hybrid_manager:
160 await self.initialize()
161 assert self.hybrid_manager is not None
163 suggestions = await self.hybrid_manager.get_autocomplete_suggestions(
164 context, cursor_position, template_name
165 )
167 return [
168 {
169 "name": item.name,
170 "type": item.type,
171 "description": item.description,
172 "signature": item.signature,
173 "adapter_source": item.adapter_source,
174 "example": item.example,
175 }
176 for item in suggestions
177 ]
179 # Fragment Management API
180 async def get_fragments_for_template(
181 self, template_name: str
182 ) -> list[dict[str, t.Any]]:
183 """Get available fragments for a template."""
184 if not self.hybrid_manager:
185 await self.initialize()
186 assert self.hybrid_manager is not None
188 fragments = await self.hybrid_manager.get_fragments_for_template(template_name)
190 return [
191 {
192 "name": fragment.name,
193 "template_path": fragment.template_path,
194 "block_name": fragment.block_name,
195 "start_line": fragment.start_line,
196 "end_line": fragment.end_line,
197 "variables": list(fragment.variables),
198 "dependencies": list(fragment.dependencies),
199 }
200 for fragment in fragments
201 ]
203 async def render_fragment(
204 self,
205 fragment_name: str,
206 context: dict[str, t.Any] | None = None,
207 template_name: str | None = None,
208 secure: bool = False,
209 ) -> str:
210 """Render a template fragment."""
211 if not self.hybrid_manager:
212 await self.initialize()
213 assert self.hybrid_manager is not None
215 return await self.hybrid_manager.render_fragment(
216 fragment_name, context, template_name, secure
217 )
219 # Enhanced Rendering API
220 async def render_template(
221 self,
222 request: Request,
223 template_name: str,
224 context: dict[str, t.Any] | None = None,
225 mode: str = "standard",
226 fragment_name: str | None = None,
227 block_name: str | None = None,
228 validate: bool = False,
229 secure: bool = False,
230 **kwargs: t.Any,
231 ) -> Response:
232 """Render template with Hybrid features."""
233 if not self.async_renderer:
234 await self.initialize()
235 assert self.async_renderer is not None
237 # Map mode string to enum
238 mode_mapping = {
239 "standard": RenderMode.STANDARD,
240 "fragment": RenderMode.FRAGMENT,
241 "block": RenderMode.BLOCK,
242 "streaming": RenderMode.STREAMING,
243 "htmx": RenderMode.HTMX,
244 }
246 RenderContext(
247 template_name=template_name,
248 context=context or {},
249 request=request,
250 mode=mode_mapping.get(mode, RenderMode.STANDARD),
251 fragment_name=fragment_name,
252 block_name=block_name,
253 validate_template=validate,
254 secure_render=secure,
255 **kwargs,
256 )
258 return await self.async_renderer.render_response(
259 request, template_name, context, **kwargs
260 )
262 # Block Rendering API
263 async def render_block(
264 self,
265 request: Request,
266 block_id: str,
267 context: dict[str, t.Any] | None = None,
268 update_mode: str = "replace",
269 target_selector: str | None = None,
270 validate: bool = False,
271 ) -> Response:
272 """Render a specific template block."""
273 if not self.block_renderer:
274 await self.initialize()
276 # Map update mode string to enum
277 update_mode_mapping = {
278 "replace": BlockUpdateMode.REPLACE,
279 "append": BlockUpdateMode.APPEND,
280 "prepend": BlockUpdateMode.PREPEND,
281 "inner": BlockUpdateMode.INNER,
282 "outer": BlockUpdateMode.OUTER,
283 "delete": BlockUpdateMode.DELETE,
284 }
286 block_request = BlockRenderRequest(
287 block_id=block_id,
288 context=context or {},
289 request=request,
290 target_selector=target_selector,
291 update_mode=update_mode_mapping.get(update_mode, BlockUpdateMode.REPLACE),
292 validate=validate,
293 )
295 result = await self.block_renderer.render_block(block_request) # type: ignore[union-attr]
297 from starlette.responses import HTMLResponse
299 return HTMLResponse(content=result.content, headers=result.htmx_headers)
301 async def render_htmx_fragment(
302 self,
303 request: Request,
304 fragment_name: str,
305 context: dict[str, t.Any] | None = None,
306 template_name: str | None = None,
307 **kwargs: t.Any,
308 ) -> Response:
309 """Render HTMX fragment with appropriate headers."""
310 if not self.async_renderer:
311 await self.initialize()
313 return await self.async_renderer.render_htmx_fragment( # type: ignore[union-attr]
314 request, fragment_name, context, template_name, **kwargs
315 )
317 # Block Management API
318 def register_htmx_block(
319 self,
320 name: str,
321 template_name: str,
322 block_name: str | None = None,
323 htmx_endpoint: str | None = None,
324 update_mode: str = "replace",
325 trigger: str = "manual",
326 auto_refresh: int | None = None,
327 **kwargs: t.Any,
328 ) -> dict[str, t.Any]:
329 """Register a block optimized for HTMX interactions."""
330 if not self.block_renderer:
331 raise RuntimeError("Block renderer not initialized")
333 from ._block_renderer import BlockTrigger, BlockUpdateMode
335 # Map string values to enums
336 update_mode_mapping = {
337 "replace": BlockUpdateMode.REPLACE,
338 "append": BlockUpdateMode.APPEND,
339 "prepend": BlockUpdateMode.PREPEND,
340 "inner": BlockUpdateMode.INNER,
341 "outer": BlockUpdateMode.OUTER,
342 "delete": BlockUpdateMode.DELETE,
343 }
345 trigger_mapping = {
346 "manual": BlockTrigger.MANUAL,
347 "auto": BlockTrigger.AUTO,
348 "lazy": BlockTrigger.LAZY,
349 "polling": BlockTrigger.POLLING,
350 "websocket": BlockTrigger.WEBSOCKET,
351 }
353 block_def = self.block_renderer.register_htmx_block(
354 name=name,
355 template_name=template_name,
356 block_name=block_name,
357 htmx_endpoint=htmx_endpoint,
358 update_mode=update_mode_mapping.get(update_mode, BlockUpdateMode.REPLACE),
359 trigger=trigger_mapping.get(trigger, BlockTrigger.MANUAL),
360 auto_refresh=auto_refresh,
361 **kwargs,
362 )
364 return {
365 "name": block_def.name,
366 "template_name": block_def.template_name,
367 "block_name": block_def.block_name,
368 "css_selector": block_def.css_selector,
369 "htmx_attrs": block_def.htmx_attrs,
370 "update_mode": block_def.update_mode.value,
371 "trigger": block_def.trigger.value,
372 }
374 async def get_block_info(self, block_id: str) -> dict[str, t.Any]:
375 """Get information about a registered block."""
376 if not self.block_renderer:
377 await self.initialize()
379 return await self.block_renderer.get_block_info(block_id) # type: ignore[union-attr]
381 def get_htmx_attributes_for_block(self, block_id: str) -> str:
382 """Get HTMX attributes string for a block."""
383 if not self.block_renderer:
384 return ""
386 return self.block_renderer.get_htmx_attributes_for_block(block_id)
388 # Performance and Monitoring API
389 async def get_performance_metrics(
390 self, template_name: str | None = None
391 ) -> dict[str, t.Any]:
392 """Get template rendering performance metrics."""
393 if not self.async_renderer:
394 await self.initialize()
396 return await self.async_renderer.get_performance_metrics(template_name) # type: ignore[union-attr]
398 def clear_caches(self) -> None:
399 """Clear all template caches."""
400 if self.hybrid_manager:
401 self.hybrid_manager.clear_caches()
403 if self.async_renderer:
404 self.async_renderer.clear_cache()
406 # Utility API
407 async def precompile_templates(self) -> dict[str, t.Any]:
408 """Precompile templates for performance optimization."""
409 if not self.hybrid_manager:
410 await self.initialize()
412 compiled = await self.hybrid_manager.precompile_templates() # type: ignore[union-attr]
413 return {name: True for name in compiled.keys()}
415 async def get_template_dependencies(self, template_name: str) -> list[str]:
416 """Get dependencies for a template."""
417 if not self.hybrid_manager:
418 await self.initialize()
420 deps = await self.hybrid_manager.get_template_dependencies(template_name) # type: ignore[union-attr]
421 return list(deps)
424# Global integration instance
425_integration_instance: HybridTemplates | None = None
428async def get_hybrid_templates() -> HybridTemplates:
429 """Get or create the global Hybrid templates integration instance."""
430 global _integration_instance
432 if _integration_instance is None:
433 _integration_instance = HybridTemplates()
434 await _integration_instance.initialize()
436 return _integration_instance
439# Convenience functions for common operations
440async def validate_template_source(
441 template_source: str,
442 template_name: str = "unknown",
443 context: dict[str, t.Any] | None = None,
444) -> dict[str, t.Any]:
445 """Validate template source code."""
446 integration = await get_hybrid_templates()
447 return await integration.validate_template(template_source, template_name, context)
450async def get_template_autocomplete(
451 context: str, cursor_position: int = 0, template_name: str = "unknown"
452) -> list[dict[str, t.Any]]:
453 """Get autocomplete suggestions for template editing."""
454 integration = await get_hybrid_templates()
455 return await integration.get_autocomplete_suggestions(
456 context, cursor_position, template_name
457 )
460async def render_htmx_block(
461 request: Request,
462 block_id: str,
463 context: dict[str, t.Any] | None = None,
464 update_mode: str = "replace",
465) -> Response:
466 """Render HTMX block with appropriate headers."""
467 integration = await get_hybrid_templates()
468 return await integration.render_block(request, block_id, context, update_mode)
471async def render_template_fragment(
472 request: Request,
473 fragment_name: str,
474 context: dict[str, t.Any] | None = None,
475 template_name: str | None = None,
476) -> Response:
477 """Render template fragment for HTMX."""
478 integration = await get_hybrid_templates()
479 return await integration.render_htmx_fragment(
480 request, fragment_name, context, template_name
481 )
484MODULE_ID = UUID("01937d8b-1234-7890-abcd-1234567890ab")
485MODULE_STATUS = AdapterStatus.STABLE
487# Register the integration
488with suppress(Exception):
489 depends.set(Templates, get_hybrid_templates)