Coverage for src/lektor_ng/admin/modules/api.py: 100%
233 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-30 13:49 +0000
1import os
2import posixpath
3from collections.abc import Callable, Iterator, Mapping
4from contextvars import ContextVar
5from dataclasses import dataclass, field
6from functools import wraps
7from typing import Any, TypeVar, cast
9import click
10import marshmallow
11import marshmallow_dataclass as mdcls
12from flask import Blueprint, Response, current_app, jsonify, make_response, request
14from lektor_ng.admin.context import LektorContext, get_lektor_context
15from lektor_ng.admin.utils import eventstream
16from lektor_ng.constants import PRIMARY_ALT
17from lektor_ng.datamodel import DataModel
18from lektor_ng.db import Record
19from lektor_ng.environment.config import Config, ServerInfo
20from lektor_ng.publisher import PublishError, publish
21from lektor_ng.utils import cleanup_path, is_valid_id
23bp = Blueprint("api", __name__, url_prefix="/admin/api")
26LEKTOR_CONFIG: ContextVar[Config] = ContextVar("lektor_config")
29@bp.url_value_preprocessor
30def pass_lektor_context(endpoint: str | None, values: dict[str, Any] | None) -> None:
31 """Pass LektorContext to each view callable in a `ctx` parameter"""
32 assert isinstance(values, dict)
33 values["ctx"] = get_lektor_context()
36class _ServerInfoField(marshmallow.fields.String):
37 def _deserialize(
38 self,
39 value: str,
40 attr: str | None,
41 data: Mapping[str, Any] | None,
42 **kwargs: Any,
43 ) -> ServerInfo:
44 lektor_config = LEKTOR_CONFIG.get()
45 server_id = super()._deserialize(value, attr, data, **kwargs)
46 server_info = lektor_config.get_server(server_id)
47 if server_info is None:
48 raise marshmallow.ValidationError("Invalid server id.")
49 return server_info
52def _is_valid_path(value: str) -> None:
53 if not cleanup_path(value) == value:
54 raise marshmallow.ValidationError("Invalid value.")
57def _is_valid_alt(value: str) -> None:
58 lektor_config = get_lektor_context().config
59 if not lektor_config.is_valid_alternative(value):
60 raise marshmallow.ValidationError("Invalid alternative.")
63# Mark types for special validation
64_PathType = mdcls.NewType("_PathType", str, validate=_is_valid_path)
65_AltType = mdcls.NewType("_AltType", str, validate=_is_valid_alt)
66_BoolType = mdcls.NewType("_BoolType", bool, truthy={1, "1"}, falsy={0, "0"})
69class _SchemaBase(marshmallow.Schema):
70 TYPE_MAPPING = {ServerInfo: _ServerInfoField}
72 class Meta:
73 unknown = marshmallow.EXCLUDE
76F = TypeVar("F", bound=Callable[..., Any])
79def _with_validated(param_type: type) -> Callable[[F], F]:
80 """Flask view decorator to validate parameters.
82 The validated parameters are placed into the ``validated`` keyword
83 arg of the decorated view.
85 If the request has a JSON body, the parameters are parsed from that.
86 Otherwise, the parameters are parsed from ``request.values``.
88 :param param_type: A dataclass which specifies the parameters.
89 """
90 schema_class = mdcls.class_schema(param_type, base_schema=_SchemaBase)
91 schema = schema_class()
93 def wrap(f: F) -> F:
94 @wraps(f)
95 def wrapper(*args: Any, **kwargs: Any) -> Response:
96 if request.method in ("POST", "PUT") and request.mimetype == "application/json":
97 data = request.get_json() or {}
98 else:
99 data = request.values
100 LEKTOR_CONFIG.set(kwargs["ctx"].config)
101 try:
102 kwargs["validated"] = schema.load(data)
103 except marshmallow.ValidationError as exc:
104 error = {
105 "title": "Invalid parameters",
106 "messages": exc.messages,
107 }
108 return make_response(jsonify(error=error), 400)
109 return f(*args, **kwargs)
111 # This cast seems necessary, atm.
112 # https://github.com/python/mypy/issues/1927
113 return cast(F, wrapper)
115 return wrap
118@dataclass
119class _PathAndAlt:
120 path: _PathType
121 alt: _AltType = PRIMARY_ALT
124@bp.route("/pathinfo")
125@_with_validated(_PathAndAlt)
126def get_path_info(validated: _PathAndAlt, ctx: LektorContext) -> Response:
127 """Returns the path segment information for a record."""
128 alt = validated.alt
129 tree_item = ctx.tree.get(validated.path)
130 segments = []
132 while tree_item is not None:
133 segments.append(
134 {
135 "id": tree_item.id,
136 "path": tree_item.path,
137 "label_i18n": tree_item.get_record_label_i18n(alt),
138 "exists": tree_item.exists,
139 "can_have_children": tree_item.can_have_children,
140 }
141 )
142 tree_item = tree_item.get_parent()
144 segments.reverse()
145 return jsonify(segments=segments)
148@bp.route("/recordinfo")
149@_with_validated(_PathAndAlt)
150def get_record_info(validated: _PathAndAlt, ctx: LektorContext) -> Response:
151 alt = validated.alt
152 tree_item = ctx.tree.get(validated.path)
154 return jsonify(
155 id=tree_item.id,
156 path=tree_item.path,
157 label_i18n=tree_item.get_record_label_i18n(alt),
158 exists=tree_item.exists,
159 is_attachment=tree_item.is_attachment,
160 attachments=[
161 {
162 "id": x.id,
163 "path": x.path,
164 "type": x.attachment_type,
165 }
166 for x in tree_item.iter_attachments()
167 ],
168 children=[
169 {
170 "id": x.id,
171 "path": x.path,
172 "label": x.id,
173 "label_i18n": x.get_record_label_i18n(alt),
174 "visible": x.is_visible,
175 }
176 for x in tree_item.iter_subpages()
177 ],
178 alts=[
179 {
180 "alt": _.id,
181 "is_primary": _.id == PRIMARY_ALT,
182 "primary_overlay": _.is_primary_overlay,
183 "name_i18n": _.name_i18n,
184 "exists": _.exists,
185 }
186 for _ in tree_item.alts.values()
187 ],
188 can_have_children=tree_item.can_have_children,
189 can_have_attachments=tree_item.can_have_attachments,
190 can_be_deleted=tree_item.can_be_deleted,
191 )
194@bp.route("/previewinfo")
195@_with_validated(_PathAndAlt)
196def get_preview_info(validated: _PathAndAlt, ctx: LektorContext) -> Response:
197 record = ctx.pad.get(validated.path, alt=validated.alt)
198 if record is None:
199 return jsonify(exists=False, url=None, is_hidden=True)
200 return jsonify(exists=True, url=record.url_path, is_hidden=record.is_hidden)
203@dataclass
204class _FindParams:
205 q: str
206 alt: _AltType = PRIMARY_ALT
207 lang: str | None = None
210@bp.route("/find", methods=["POST"])
211@_with_validated(_FindParams)
212def find(validated: _FindParams, ctx: LektorContext) -> Response:
213 lang = validated.lang or current_app.config.get("lektor.ui_lang", "en")
214 return jsonify(results=ctx.builder.find_files(validated.q, alt=validated.alt, lang=lang))
217@bp.route("/browsefs", methods=["POST"])
218@_with_validated(_PathAndAlt)
219def browsefs(validated: _PathAndAlt, ctx: LektorContext) -> Response:
220 record = ctx.pad.get(validated.path, alt=validated.alt)
221 okay = False
222 if record is not None:
223 if record.is_attachment:
224 fn = record.attachment_filename
225 else:
226 fn = record.source_filename
227 if os.path.exists(fn):
228 click.launch(fn, locate=True)
229 okay = True
230 return jsonify(okay=okay)
233@dataclass
234class _UrlPath:
235 url_path: str
238@bp.route("/matchurl")
239@_with_validated(_UrlPath)
240def match_url(validated: _UrlPath, ctx: LektorContext) -> Response:
241 """Find the Record that corresponds to a URL.
243 This is used by the admin UI to find the db record that corresponds
244 to a page when the preview iframe is navigated.
245 """
246 record = ctx.pad.resolve_url_path(validated.url_path, alt_fallback=False)
247 if not isinstance(record, Record):
248 return jsonify(exists=False, path=None, alt=None)
249 return jsonify(exists=True, path=record["_path"], alt=record["_alt"])
252@bp.route("/rawrecord")
253@_with_validated(_PathAndAlt)
254def get_raw_record(validated: _PathAndAlt, ctx: LektorContext) -> Response:
255 ts = ctx.tree.edit(validated.path, alt=validated.alt)
256 return jsonify(ts.to_json())
259@bp.route("/newrecord")
260@_with_validated(_PathAndAlt)
261def get_new_record_info(validated: _PathAndAlt, ctx: LektorContext) -> Response:
262 pad = ctx.pad
263 alt = validated.alt
264 tree_item = ctx.tree.get(validated.path)
266 def describe_model(model: DataModel) -> dict[str, Any]:
267 primary_field = None
268 if model.primary_field is not None:
269 f = model.field_map.get(model.primary_field)
270 if f is not None:
271 primary_field = f.to_json(pad)
272 return {
273 "id": model.id,
274 "name": model.name,
275 "name_i18n": model.name_i18n,
276 "primary_field": primary_field,
277 }
279 implied_model = tree_item.implied_child_datamodel
280 label_i18n = tree_item.get_record_label_i18n(alt)
281 return jsonify(
282 {
283 "label_i18n": label_i18n,
284 "label": label_i18n["en"],
285 "can_have_children": tree_item.can_have_children,
286 "implied_model": implied_model,
287 "available_models": {
288 k: describe_model(v) for k, v in pad.db.datamodels.items() if not v.hidden or k == implied_model
289 },
290 }
291 )
294@bp.route("/newattachment")
295@_with_validated(_PathAndAlt)
296def get_new_attachment_info(validated: _PathAndAlt, ctx: LektorContext) -> Response:
297 tree_item = ctx.tree.get(validated.path)
298 label_i18n = tree_item.get_record_label_i18n(validated.alt)
299 return jsonify(
300 {
301 "can_upload": tree_item.can_have_attachments,
302 "label_i18n": label_i18n,
303 "label": label_i18n["en"],
304 }
305 )
308@bp.route("/newattachment", methods=["POST"])
309@_with_validated(_PathAndAlt)
310def upload_new_attachments(validated: _PathAndAlt, ctx: LektorContext) -> Response:
311 ts = ctx.tree.edit(validated.path, alt=validated.alt)
312 if not ts.exists or ts.is_attachment:
313 return jsonify({"bad_upload": True})
315 buckets = []
316 for file in request.files.getlist("file"):
317 stored_filename = ts.add_attachment(file.filename, file)
318 buckets.append({"original_filename": file.filename, "stored_filename": stored_filename})
320 return jsonify(
321 {
322 "bad_upload": False,
323 "path": validated.path,
324 "buckets": buckets,
325 }
326 )
329@dataclass
330class _NewRecordParams:
331 id: str
332 model: str | None
333 data: dict[str, str | None]
334 path: _PathType
335 alt: _AltType = PRIMARY_ALT
338@bp.route("/newrecord", methods=["POST"])
339@_with_validated(_NewRecordParams)
340def add_new_record(validated: _NewRecordParams, ctx: LektorContext) -> Response:
341 exists = False
343 if not is_valid_id(validated.id):
344 return jsonify(valid_id=False, exists=False, path=None)
346 path = posixpath.join(validated.path, validated.id)
348 ts = ctx.tree.edit(path, datamodel=validated.model, alt=validated.alt)
349 with ts:
350 if ts.exists:
351 exists = True
352 else:
353 ts.data.update(validated.data)
355 return jsonify({"valid_id": True, "exists": exists, "path": path})
358@dataclass
359class _DeleteRecordParams:
360 delete_master: _BoolType
361 path: _PathType
362 alt: _AltType = PRIMARY_ALT
365@bp.route("/deleterecord", methods=["POST"])
366@_with_validated(_DeleteRecordParams)
367def delete_record(validated: _DeleteRecordParams, ctx: LektorContext) -> Response:
368 if validated.path != "/":
369 ts = ctx.tree.edit(validated.path, alt=validated.alt)
370 with ts:
371 ts.delete(delete_master=validated.delete_master)
372 return jsonify(okay=True)
375@dataclass
376class _UpdateRawRecordParams:
377 data: dict[str, str | None]
378 path: _PathType
379 alt: _AltType = PRIMARY_ALT
382@bp.route("/rawrecord", methods=["PUT"])
383@_with_validated(_UpdateRawRecordParams)
384def update_raw_record(validated: _UpdateRawRecordParams, ctx: LektorContext) -> Response:
385 ts = ctx.tree.edit(validated.path, alt=validated.alt)
386 with ts:
387 ts.data.update(validated.data)
388 return jsonify(path=ts.path)
391@bp.route("/servers")
392def get_servers(ctx: LektorContext) -> Response:
393 servers = ctx.config.get_servers(public=True)
394 return jsonify(servers=sorted([x.to_json() for x in servers.values()], key=lambda x: x["name"].lower()))
397@bp.route("/build", methods=["POST"])
398def trigger_build(ctx: LektorContext) -> Response:
399 builder = ctx.builder
400 builder.build_all()
401 builder.prune()
402 return jsonify(okay=True)
405@bp.route("/clean", methods=["POST"])
406def trigger_clean(ctx: LektorContext) -> Response:
407 builder = ctx.builder
408 builder.prune(all=True)
409 builder.touch_site_config()
410 return jsonify(okay=True)
413@dataclass
414class _PublishBuildParams:
415 server_info: ServerInfo = field(metadata={"data_key": "server"})
418@bp.route("/publish")
419@_with_validated(_PublishBuildParams)
420def publish_build(validated: _PublishBuildParams, ctx: LektorContext) -> Response:
421 @eventstream
422 def generator() -> Iterator[dict[str, str]]:
423 try:
424 event_iter = (
425 publish(
426 ctx.env,
427 validated.server_info.target,
428 ctx.output_path,
429 server_info=validated.server_info,
430 )
431 or ()
432 )
433 for event in event_iter:
434 yield {"msg": event}
435 except PublishError as exc:
436 yield {"msg": f"Error: {exc}"}
438 return generator()
441@bp.route("/ping")
442def ping(ctx: LektorContext) -> Response:
443 return jsonify(project_id=ctx.project_id, okay=True)