Coverage for src/local_deep_research/web/template_config.py: 89%
84 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Shared Jinja2 templates instance.
4This module exists to break the circular import between fastapi_app.py
5and router modules. Routers import templates from here instead of
6from fastapi_app.py.
7"""
9from importlib import resources as importlib_resources
10from pathlib import Path
11from typing import Any
13from fastapi.templating import Jinja2Templates
14from starlette.requests import Request
15from starlette.responses import Response
18# Starlette's TemplateResponse(request, name, context, status_code, headers,
19# media_type, background) parameters, in order.
20_TEMPLATE_RESPONSE_ARGS = (
21 "request",
22 "name",
23 "context",
24 "status_code",
25 "headers",
26 "media_type",
27 "background",
28)
31class _LDRTemplates(Jinja2Templates):
32 """Custom Jinja2Templates that injects CSRF token + flash messages
33 into every TemplateResponse, so routes can use templates.TemplateResponse
34 directly without losing CSRF/flash support.
35 """
37 def TemplateResponse( # noqa: N802 — match parent API
38 self,
39 *args: Any,
40 **kwargs: Any,
41 ) -> Response:
42 # Normalise: support both positional and keyword forms.
43 # Jinja2Templates.TemplateResponse(request, name, context, ...) OR
44 # Jinja2Templates.TemplateResponse(name, context, request=, ...) (legacy) OR
45 # Jinja2Templates.TemplateResponse(request=, name=, context=, ...)
46 # Starlette's own form is folded into keywords first, so the context
47 # the defaults go into is the one rendered: read from kwargs only, a
48 # positional context was replaced by a fresh {} and then passed a
49 # second time as context=, a TypeError.
50 if args and isinstance(args[0], Request):
51 for key, value in zip(_TEMPLATE_RESPONSE_ARGS, args):
52 if key in kwargs:
53 raise TypeError(
54 f"TemplateResponse() got multiple values for argument {key!r}"
55 )
56 kwargs[key] = value
57 args = args[len(_TEMPLATE_RESPONSE_ARGS) :]
58 context = kwargs.get("context") or {}
59 request = kwargs.get("request") or context.get("request")
61 if request is None and args: 61 ↛ 63line 61 didn't jump to line 63 because the condition on line 61 was never true
62 # First positional could be Request
63 if isinstance(args[0], Request):
64 request = args[0]
66 session = getattr(request, "session", None) if request else None
67 # Session is a Starlette mapping (dict-like); guard conservatively
68 # — the old check `session.__class__.get` crashed when session was
69 # None (NoneType has no `.get`) and was vacuously true otherwise.
70 if request is not None and hasattr(session, "get"): 70 ↛ 98line 70 didn't jump to line 98 because the condition on line 70 was always true
71 # Inject session so templates can do {{ session.username }}.
72 # Every base template reads session.username for the top bar.
73 context.setdefault("session", session)
74 from .dependencies.csrf import get_auth_context
76 context["auth_context"] = get_auth_context(session)
78 # Inject CSRF token if not already present
79 if "csrf_token" not in context:
80 from .dependencies.csrf import generate_csrf_token
82 token = generate_csrf_token(request) # gitleaks:allow
83 context["csrf_token"] = lambda: token
85 # Inject flash messages if not already present
86 if "get_flashed_messages" not in context:
87 from .dependencies.flash import get_flashed_messages
89 flashes = get_flashed_messages(request, with_categories=True)
90 context["get_flashed_messages"] = lambda with_categories=False: (
91 flashes if with_categories else [msg for _, msg in flashes]
92 )
94 kwargs["context"] = context
96 # Inject frontend constants used by base.html (ports Flask's
97 # inject_frontend_constants context processor from app_factory.py).
98 if "research_status_enum" not in context: 98 ↛ 148line 98 didn't jump to line 148 because the condition on line 98 was always true
99 from ..constants import ResearchStatus
101 terminal = [
102 ResearchStatus.COMPLETED,
103 ResearchStatus.SUSPENDED,
104 ResearchStatus.FAILED,
105 ResearchStatus.ERROR,
106 ResearchStatus.CANCELLED,
107 ]
108 context["research_status_enum"] = {
109 m.name: m.value for m in ResearchStatus
110 }
111 context["research_terminal_states"] = [str(s) for s in terminal]
112 from ..constants import (
113 DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP,
114 DEFAULT_LOCAL_SEARCH_CHUNK_SIZE,
115 DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC,
116 DEFAULT_LOCAL_SEARCH_INDEX_TYPE,
117 DEFAULT_LOCAL_SEARCH_MODEL,
118 DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
119 DEFAULT_LOCAL_SEARCH_PROVIDER,
120 DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE,
121 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS,
122 HISTORY_LOGS_DEFAULT_LIMIT,
123 HISTORY_LOGS_HARD_CAP,
124 )
126 context["log_limits"] = {
127 "default": HISTORY_LOGS_DEFAULT_LIMIT,
128 "hard_cap": HISTORY_LOGS_HARD_CAP,
129 }
130 context["local_search_defaults"] = {
131 "provider": DEFAULT_LOCAL_SEARCH_PROVIDER,
132 "model": DEFAULT_LOCAL_SEARCH_MODEL,
133 "chunk_size": DEFAULT_LOCAL_SEARCH_CHUNK_SIZE,
134 "chunk_overlap": DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP,
135 "splitter_type": DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE,
136 "text_separators": list(DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS),
137 "distance_metric": DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC,
138 "normalize_vectors": DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
139 "index_type": DEFAULT_LOCAL_SEARCH_INDEX_TYPE,
140 }
141 kwargs["context"] = context
143 # Inject the app version for every render (main did this via
144 # render_template_with_defaults' `version=__version__`). The sidebar
145 # renders a version badge linking to the matching release tag; without
146 # injection `version` is Undefined and every page shows an empty badge
147 # pointing at .../releases/tag/v.
148 if "version" not in context: 148 ↛ 160line 148 didn't jump to line 160 because the condition on line 148 was always true
149 from ..__version__ import __version__
151 context["version"] = __version__
152 kwargs["context"] = context
154 # Inject has_encryption for every render (main did this via
155 # render_template_with_defaults). settings_dashboard.html gates its
156 # "Database encryption is not available" security warning on
157 # `{% if not has_encryption %}`; without injection the variable is
158 # Undefined → the warning renders on every page load including
159 # fully encrypted installs, training users to ignore it.
160 if "has_encryption" not in context:
161 from ..database.encrypted_db import db_manager
163 context["has_encryption"] = db_manager.has_encryption
164 kwargs["context"] = context
166 # Inject filesystem_pdf_storage_allowed for every render so that
167 # templates (such as storage_mode_modal.html) can gate the unencrypted
168 # filesystem storage option without requiring each route to remember
169 # passing it.
170 if "filesystem_pdf_storage_allowed" not in context: 170 ↛ 187line 170 didn't jump to line 187 because the condition on line 170 was always true
171 from ..research_library.services.pdf_storage_manager import (
172 filesystem_pdf_storage_allowed,
173 )
175 context["filesystem_pdf_storage_allowed"] = (
176 filesystem_pdf_storage_allowed()
177 )
178 kwargs["context"] = context
180 # Inject the user's saved egress scope so base.html renders it onto
181 # <body data-scope=…> on EVERY page. Previously only routes going
182 # through dependencies/template_helpers.render_template got this;
183 # pages using templates.TemplateResponse directly (history, metrics,
184 # …) silently fell back to the module-level Jinja default, so the
185 # scope-aware CSS cues vanished on those pages. Fail open to the
186 # default — never crash a page render over a styling cue.
187 if "egress_scope" not in context:
188 from ..security.egress.policy import DEFAULT_EGRESS_SCOPE
190 scope = DEFAULT_EGRESS_SCOPE
191 try:
192 username = (
193 session.get("username") if hasattr(session, "get") else None
194 )
195 if username:
196 from ..database.session_context import (
197 get_user_db_session,
198 )
199 from ..utilities.db_utils import get_settings_manager
201 with get_user_db_session(username) as db_session:
202 if db_session: 202 ↛ 230line 202 didn't jump to line 230
203 sm = get_settings_manager(db_session, username)
204 scope = (
205 sm.get_setting(
206 "policy.egress_scope",
207 DEFAULT_EGRESS_SCOPE,
208 )
209 or DEFAULT_EGRESS_SCOPE
210 )
211 # Canonicalise before it reaches the template: this collapses case
212 # and whitespace and maps invalid / retired values to the
213 # protective default. Without it a stored "STRICT" renders
214 # data-scope="STRICT", and base.html's body[data-scope="strict"]
215 # selector -- attribute values are case-sensitive -- silently stops
216 # matching, so the strict-scope visual cue disappears. Lost in the
217 # port; main applied it at app_factory.py:585.
218 from ..security.egress.policy import (
219 effective_scope_for_display,
220 )
222 scope = effective_scope_for_display(scope)
223 except Exception:
224 from loguru import logger
226 logger.debug(
227 "Failed to read egress scope for template context",
228 exc_info=True,
229 )
230 context["egress_scope"] = scope
231 kwargs["context"] = context
233 return super().TemplateResponse(*args, **kwargs)
236try:
237 _PACKAGE_DIR = importlib_resources.files("local_deep_research") / "web"
238 with importlib_resources.as_file(_PACKAGE_DIR) as _pkg:
239 TEMPLATE_DIR = (_pkg / "templates").as_posix()
240 STATIC_DIR = (_pkg / "static").as_posix()
241except Exception:
242 TEMPLATE_DIR = str(Path("templates").resolve())
243 STATIC_DIR = str(Path("static").resolve())
245templates = _LDRTemplates(directory=TEMPLATE_DIR)