Coverage for src/lektor_ng/db.py: 89%
1294 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-08 00:39 +0000
1# pylint: disable=too-many-lines
2from __future__ import annotations
4import errno
5import functools
6import hashlib
7import operator
8import os
9import posixpath
10from collections import OrderedDict
11from datetime import timedelta
12from functools import total_ordering
13from itertools import chain, islice
14from operator import methodcaller
15from pathlib import Path
16from typing import TYPE_CHECKING
17from urllib.parse import urljoin
19from jinja2 import Undefined, is_undefined
20from jinja2.exceptions import UndefinedError
21from jinja2.utils import LRUCache
22from werkzeug.utils import cached_property
24from lektor_ng import metaformat
25from lektor_ng.assets import get_asset_root
26from lektor_ng.constants import PRIMARY_ALT
27from lektor_ng.context import Context, get_ctx
28from lektor_ng.databags import Databags
29from lektor_ng.datamodel import load_datamodels, load_flowblocks
30from lektor_ng.editor import make_editor_session
31from lektor_ng.filecontents import FileContents
32from lektor_ng.imagetools import (
33 ThumbnailMode,
34 get_image_info,
35 make_image_thumbnail,
36 read_exif,
37)
38from lektor_ng.sourceobj import DBSourceObject, VirtualSourceObject
39from lektor_ng.utils import (
40 cleanup_path,
41 cleanup_url_path,
42 deprecated,
43 fs_enc,
44 locate_executable,
45 make_relative_url,
46 sort_normalize_string,
47 split_virtual_path,
48 untrusted_to_os_path,
49)
50from lektor_ng.videotools import get_video_info, make_video_thumbnail
52if TYPE_CHECKING:
53 from lektor.environment import Environment
54 from lektor.environment.config import Config
56# pylint: disable=no-member
59def get_alts(source=None, fallback=False):
60 """Given a source this returns the list of all alts that the source
61 exists as. It does not include fallbacks unless `fallback` is passed.
62 If no source is provided all configured alts are returned. If alts are
63 not configured at all, the return value is an empty list.
64 """
65 if source is None:
66 ctx = get_ctx()
67 if ctx is None:
68 raise RuntimeError("This function requires the context to be supplied.")
69 pad = ctx.pad
70 else:
71 pad = source.pad
72 alts = list(pad.config.iter_alternatives())
73 if alts == [PRIMARY_ALT]:
74 return []
76 rv = alts
78 # If a source is provided and it's not virtual, we look up all alts
79 # of the path on the pad to figure out which records exist.
80 if source is not None and "@" not in source.path:
81 rv = []
82 for alt in alts:
83 if pad.alt_exists(source.path, alt=alt, fallback=fallback):
84 rv.append(alt)
86 return rv
89def _require_ctx(record):
90 ctx = get_ctx()
91 if ctx is None:
92 raise RuntimeError("This operation requires a context but none was on the stack.")
93 if ctx.pad is not record.pad:
94 raise RuntimeError("The context on the stack does not match the pad of the record.")
95 return ctx
98@total_ordering
99class _CmpHelper:
100 def __init__(self, value, reverse):
101 self.value = value
102 self.reverse = reverse
104 @staticmethod
105 def coerce(a, b):
106 if isinstance(a, str) and isinstance(b, str):
107 return sort_normalize_string(a), sort_normalize_string(b)
108 if type(a) is type(b):
109 return a, b
110 if isinstance(a, Undefined) or isinstance(b, Undefined):
111 if isinstance(a, Undefined):
112 a = None
113 if isinstance(b, Undefined):
114 b = None
115 return a, b
116 if isinstance(a, (int, float)):
117 try:
118 return a, type(a)(b)
119 except (ValueError, TypeError, OverflowError):
120 pass
121 if isinstance(b, (int, float)):
122 try:
123 return type(b)(a), b
124 except (ValueError, TypeError, OverflowError):
125 pass
126 return a, b
128 def __eq__(self, other):
129 a, b = self.coerce(self.value, other.value)
130 return a == b
132 def __lt__(self, other):
133 a, b = self.coerce(self.value, other.value)
134 try:
135 if self.reverse:
136 return b < a
137 return a < b
138 except TypeError:
139 # Put None at the beginning if reversed, else at the end.
140 if self.reverse:
141 return a is not None
142 return a is None
145def _auto_wrap_expr(value):
146 if isinstance(value, Expression):
147 return value
148 return _Literal(value)
151def save_eval(filter, record):
152 try:
153 return filter.__eval__(record)
154 except UndefinedError as e:
155 return Undefined(e.message)
158class Expression:
159 def __eval__(self, record):
160 # pylint: disable=no-self-use
161 return record
163 def __eq__(self, other):
164 return _BinExpr(self, _auto_wrap_expr(other), operator.eq)
166 def __ne__(self, other):
167 return _BinExpr(self, _auto_wrap_expr(other), operator.ne)
169 def __and__(self, other):
170 return _BinExpr(self, _auto_wrap_expr(other), operator.and_)
172 def __or__(self, other):
173 return _BinExpr(self, _auto_wrap_expr(other), operator.or_)
175 def __gt__(self, other):
176 return _BinExpr(self, _auto_wrap_expr(other), operator.gt)
178 def __ge__(self, other):
179 return _BinExpr(self, _auto_wrap_expr(other), operator.ge)
181 def __lt__(self, other):
182 return _BinExpr(self, _auto_wrap_expr(other), operator.lt)
184 def __le__(self, other):
185 return _BinExpr(self, _auto_wrap_expr(other), operator.le)
187 def contains(self, item):
188 return _ContainmentExpr(self, _auto_wrap_expr(item))
190 def startswith(self, other):
191 return _BinExpr(
192 self,
193 _auto_wrap_expr(other),
194 lambda a, b: str(a).lower().startswith(str(b).lower()),
195 )
197 def endswith(self, other):
198 return _BinExpr(
199 self,
200 _auto_wrap_expr(other),
201 lambda a, b: str(a).lower().endswith(str(b).lower()),
202 )
204 def startswith_cs(self, other):
205 return _BinExpr(
206 self,
207 _auto_wrap_expr(other),
208 lambda a, b: str(a).startswith(str(b)),
209 )
211 def endswith_cs(self, other):
212 return _BinExpr(
213 self,
214 _auto_wrap_expr(other),
215 lambda a, b: str(a).endswith(str(b)),
216 )
218 def false(self):
219 return _IsBoolExpr(self, False)
221 def true(self):
222 return _IsBoolExpr(self, True)
225# Query helpers for the template engine
226setattr(Expression, "and", lambda x, o: x & o)
227setattr(Expression, "or", lambda x, o: x | o)
230class _CallbackExpr(Expression):
231 def __init__(self, func):
232 self.func = func
234 def __eval__(self, record):
235 return self.func(record)
238class _IsBoolExpr(Expression):
239 def __init__(self, expr, true):
240 self.__expr = expr
241 self.__true = true
243 def __eval__(self, record):
244 val = self.__expr.__eval__(record)
245 return (not is_undefined(val) and val not in (None, 0, False, "")) == self.__true
248class _Literal(Expression):
249 def __init__(self, value):
250 self.__value = value
252 def __eval__(self, record):
253 return self.__value
256class _BinExpr(Expression):
257 def __init__(self, left, right, op):
258 self.__left = left
259 self.__right = right
260 self.__op = op
262 def __eval__(self, record):
263 return self.__op(self.__left.__eval__(record), self.__right.__eval__(record))
266class _ContainmentExpr(Expression):
267 def __init__(self, seq, item):
268 self.__seq = seq
269 self.__item = item
271 def __eval__(self, record):
272 seq = self.__seq.__eval__(record)
273 item = self.__item.__eval__(record)
274 if isinstance(item, Record):
275 item = item["_id"]
276 return item in seq
279class _RecordQueryField(Expression):
280 def __init__(self, field):
281 self.__field = field
283 def __eval__(self, record):
284 try:
285 return record[self.__field]
286 except KeyError:
287 return Undefined(obj=record, name=self.__field)
290class _RecordQueryProxy:
291 def __getattr__(self, name):
292 if name[:2] != "__":
293 return _RecordQueryField(name)
294 raise AttributeError(name)
296 def __getitem__(self, name):
297 try:
298 return self.__getattr__(name)
299 except AttributeError as error:
300 raise KeyError(name) from error
303F = _RecordQueryProxy()
306class Record(DBSourceObject):
307 source_classification = "record"
308 supports_pagination = False
310 def __init__(self, pad, data, page_num=None):
311 super().__init__(pad)
312 self._data = data
313 self._bound_data = {}
314 if page_num is not None and not self.supports_pagination:
315 raise RuntimeError(f"{self.__class__.__name__} does not support pagination")
316 self.page_num = page_num
318 @property
319 def record(self):
320 return self
322 @property
323 def datamodel(self):
324 """Returns the data model for this record."""
325 try:
326 return self.pad.db.datamodels[self._data["_model"]]
327 except LookupError:
328 # If we cannot find the model we fall back to the default one.
329 return self.pad.db.default_model
331 @property
332 def alt(self):
333 """Returns the alt of this source object."""
334 return self._data["_alt"]
336 @property
337 def is_hidden(self) -> bool:
338 """Indicates whether a record is hidden.
340 Artifacts are not built for hidden objects. Also, by default, hidden records
341 are not included in `Query` results.
342 """
343 hidden = self._data["_hidden"]
344 if not is_undefined(hidden):
345 return hidden
346 return self._is_hidden_by_parent_config()
348 def _is_hidden_by_parent_config(self) -> bool:
349 # Records may be implicitly hidden by their parents' configuration. The details
350 # depend on record type.
352 # The Page and Attachment subclasses provide concrete implementations for this
353 # method.
354 raise NotImplementedError
356 @property
357 def is_discoverable(self) -> bool:
358 """Indicates whether the page is discoverable without knowing the URL."""
359 return self._data["_discoverable"]
361 @cached_property
362 def pagination(self):
363 """Returns the pagination controller for the record."""
364 if not self.supports_pagination:
365 raise AttributeError()
366 return self.datamodel.pagination_config.get_pagination_controller(self)
368 @cached_property
369 @deprecated(version="3.4.0", stacklevel=2)
370 def contents(self):
371 return FileContents(self.source_filename)
373 def get_fallback_record_label(self, lang):
374 if not self["_id"]:
375 return "(Index)"
376 return self["_id"].replace("-", " ").replace("_", " ").title()
378 def get_record_label_i18n(self):
379 rv = {}
380 for lang in self.datamodel.label_i18n or {}:
381 label = self.datamodel.format_record_label(self, lang)
382 if not label:
383 label = self.get_fallback_record_label(lang)
384 rv[lang] = label
385 # Fill in english if missing
386 if "en" not in rv:
387 rv["en"] = self.get_fallback_record_label("en")
388 return rv
390 @property
391 def record_label(self):
392 return (self.get_record_label_i18n() or {}).get("en")
394 @property
395 def url_path(self):
396 # This is redundant (it's the same as the inherited
397 # SourceObject.url_path) but is here to silence
398 # pylint ("W0223: Method 'url_path' is abstract in class
399 # 'SourceObject' but is not overridden (abstract-method)"),
400 # as well as to document that Record is an abstract class.
401 raise NotImplementedError()
403 def _get_clean_url_path(self):
404 """The "clean" URL path, before modification to account for alt and
405 page_num and without any leading '/'
406 """
407 bits = [self["_slug"]]
408 parent = self.parent
409 while parent is not None:
410 slug = parent["_slug"]
411 head, sep, tail = slug.rpartition("/")
412 if "." in tail:
413 # https://www.getlektor.com/docs/content/urls/#content-below-dotted-slugs
414 slug = head + sep + f"_{tail}"
415 bits.append(slug)
416 parent = parent.parent
417 return "/".join(reversed(bits)).strip("/")
419 def _get_url_path(self, alt=None):
420 """The target path where the record should end up, after adding prefix/suffix
421 for the specified alt (but before accounting for any page_num).
423 Note that some types of records (Attachments), are only
424 created for the primary alternative.
425 """
426 clean_path = self._get_clean_url_path()
427 config = self.pad.config
428 if config.primary_alternative:
429 # alternatives are configured
430 if alt is None:
431 alt = config.primary_alternative
432 prefix, suffix = config.get_alternative_url_span(alt)
433 # XXX: 404.html with suffix -de becomes 404.html-de but should
434 # actually become 404-de.html
435 clean_path = prefix.lstrip("/") + clean_path + suffix.rstrip("/")
436 return "/" + clean_path.rstrip("/")
438 @property
439 def path(self):
440 return self._data["_path"]
442 def get_sort_key(self, fields):
443 """Returns a sort key for the given field specifications specific
444 for the data in the record.
445 """
446 rv = [None] * len(fields)
447 for idx, field in enumerate(fields):
448 if field[:1] == "-":
449 field = field[1:]
450 reverse = True
451 else:
452 field = field.lstrip("+")
453 reverse = False
454 try:
455 value = self[field]
456 except KeyError:
457 value = None
458 rv[idx] = _CmpHelper(value, reverse)
459 return rv
461 def __contains__(self, name):
462 return name in self._data and not is_undefined(self._data[name])
464 def __getitem__(self, name):
465 rv = self._bound_data.get(name, Ellipsis)
466 if rv is not Ellipsis:
467 return rv
468 rv = self._data[name]
469 if hasattr(rv, "__get__"):
470 rv = rv.__get__(self)
471 self._bound_data[name] = rv
472 return rv
474 def __repr__(self):
475 bits = [
476 f"model={self._data['_model']!r}",
477 f"path={self._data['_path']!r}",
478 ]
479 if self.alt != PRIMARY_ALT:
480 bits.append(f"alt={self.alt!r}")
481 if self.page_num is not None:
482 bits.append(f"page_num={self.page_num!r}")
483 return f"<{self.__class__.__name__} {' '.join(bits)}>"
486class Siblings(VirtualSourceObject): # pylint: disable=abstract-method
487 def __init__(self, record, prev_page, next_page):
488 """Virtual source representing previous and next sibling of 'record'."""
489 VirtualSourceObject.__init__(self, record)
490 self._path = record.path + "@siblings"
491 self._prev_page = prev_page
492 self._next_page = next_page
494 @property
495 def path(self):
496 # Used as a key in Context.referenced_virtual_dependencies.
497 return self._path
499 @property
500 def prev_page(self):
501 return self._prev_page
503 @property
504 def next_page(self):
505 return self._next_page
507 def iter_source_filenames(self):
508 for page in self._prev_page, self._next_page:
509 if page:
510 yield page.source_filename
512 def _file_infos(self, path_cache):
513 for page in self._prev_page, self._next_page:
514 if page:
515 yield path_cache.get_file_info(page.source_filename)
517 def get_mtime(self, path_cache):
518 mtimes = [i.mtime for i in self._file_infos(path_cache)]
519 return max(mtimes) if mtimes else None
521 def get_checksum(self, path_cache):
522 sums = "|".join(i.filename_and_checksum for i in self._file_infos(path_cache))
524 return sums or None
527def siblings_resolver(node, url_path):
528 return node.get_siblings()
531class Page(Record):
532 """This represents a loaded record."""
534 is_attachment = False
535 supports_pagination = True
537 @cached_property
538 def path(self):
539 rv = self._data["_path"]
540 if self.page_num is not None:
541 rv = f"{rv}@{self.page_num}"
542 return rv
544 @cached_property
545 def record(self):
546 if self.page_num is None:
547 return self
548 return self.pad.get(
549 self._data["_path"],
550 persist=self.pad.cache.is_persistent(self),
551 alt=self.alt,
552 )
554 def iter_source_filenames(self):
555 fs_path = self.pad.db.to_fs_path(self._data["_path"])
556 if self.alt != PRIMARY_ALT:
557 yield os.path.join(fs_path, f"contents+{self.alt}.lr")
558 yield os.path.join(fs_path, "contents.lr")
560 @property
561 def url_path(self):
562 pg = self.datamodel.pagination_config
563 path = self._get_url_path(self.alt)
564 _, _, last_part = path.rpartition("/")
565 if not pg.enabled:
566 if "." in last_part:
567 return path
568 return path.rstrip("/") + "/"
569 if "." in last_part:
570 raise RuntimeError("When file extension is provided pagination cannot be used.")
571 # pagination is enabled
572 if self.page_num in (1, None):
573 return path.rstrip("/") + "/"
574 return f"{path.rstrip('/')}/{pg.url_suffix.strip('/')}/{self.page_num:d}/"
576 @property
577 def url_content_path(self):
578 """URL path to the directory that contains children of this record."""
579 url_path = self.url_path
580 if url_path.endswith("/"):
581 return url_path
582 # See https://www.getlektor.com/docs/content/urls/#content-below-dotted-slugs
583 head, sep, tail = url_path.rpartition("/")
584 assert "." in tail
585 return f"{head}{sep}_{tail}/"
587 def resolve_url_path(self, url_path):
588 pg = self.datamodel.pagination_config
590 # If we hit the end of the url path, then we found our target.
591 # However if pagination is enabled we want to resolve the first
592 # page instead of the unpaginated version.
593 if not url_path:
594 if pg.enabled and self.page_num is None:
595 return pg.get_record_for_page(self, 1)
596 return self
598 # Try to resolve the correctly paginated version here.
599 if pg.enabled:
600 rv = pg.match_pagination(self, url_path)
601 if rv is not None:
602 return rv
604 # When we resolve URLs we also want to be able to explicitly
605 # target undiscoverable pages. Those who know the URL are
606 # rewarded.
608 # We also want to resolve hidden children
609 # here. Pad.resolve_url_path() is where the check for hidden
610 # records is done.
611 q = self.children.include_undiscoverable(True).include_hidden(True)
613 for idx in range(len(url_path)):
614 piece = "/".join(url_path[: idx + 1])
615 child = q.filter(F._slug == piece).first()
616 if child is None:
617 attachment = self.attachments.filter(F._slug == piece).first()
618 if attachment is None:
619 obj = self.pad.db.env.resolve_custom_url_path(self, url_path)
620 if obj is None:
621 continue
622 node = obj
623 else:
624 node = attachment
625 else:
626 node = child
628 rv = node.resolve_url_path(url_path[idx + 1 :])
629 if rv is not None:
630 return rv
632 if len(url_path) == 1 and url_path[0] == "index.html":
633 if pg.enabled or "." not in self["_slug"]:
634 # This page renders to an index.html. Its .url_path method returns
635 # a path ending with '/'. Accept explicit "/index.html" when resolving.
636 #
637 # FIXME: the code for Record (and subclass) .url_path and .resolve_url
638 # could use some cleanup, especially where it deals with
639 # slugs that contain '.'s.
640 return self
642 return None
644 @cached_property
645 def parent(self):
646 """The parent of the record."""
647 this_path = self._data["_path"]
648 parent_path = posixpath.dirname(this_path)
649 if parent_path != this_path:
650 return self.pad.get(parent_path, persist=self.pad.cache.is_persistent(self), alt=self.alt)
651 return None
653 def _is_hidden_by_parent_config(self) -> bool:
654 # For Pages, If an explicit value for the _hidden field is not set, the value of
655 # the hidden option in the child configuration section of the parent's datamodel
656 # is checked. If that, too, is not set, then pages inherit the hidden status of
657 # their parent.
658 parent = self.parent
659 if parent is None:
660 return False
661 hidden_children = parent.datamodel.child_config.hidden
662 if hidden_children is not None:
663 return hidden_children
664 return parent.is_hidden
666 @property
667 def children(self):
668 """A query over all children that are not hidden or undiscoverable.
669 want undiscoverable then use ``children.include_undiscoverable(True)``.
670 """
671 repl_query = self.datamodel.get_child_replacements(self)
672 if repl_query is not None:
673 return repl_query.include_undiscoverable(False)
674 return Query(path=self._data["_path"], pad=self.pad, alt=self.alt)
676 @property
677 def attachments(self):
678 """Returns a query for the attachments of this record."""
679 return AttachmentsQuery(path=self._data["_path"], pad=self.pad, alt=self.alt)
681 def has_prev(self):
682 return self.get_siblings().prev_page is not None
684 def has_next(self):
685 return self.get_siblings().next_page is not None
687 def get_siblings(self):
688 """The next and previous children of this page's parent.
690 Uses parent's pagination query, if any, else parent's "children" config.
691 """
692 siblings = Siblings(self, *self._siblings)
693 ctx = get_ctx()
694 if ctx:
695 ctx.pad.db.track_record_dependency(siblings)
696 return siblings
698 @cached_property
699 def _siblings(self):
700 parent = self.parent
701 pagination_enabled = parent.datamodel.pagination_config.enabled
703 # Don't track dependencies for this part.
704 with Context(pad=self.pad):
705 if pagination_enabled:
706 pagination = parent.pagination
707 siblings = list(pagination.config.get_pagination_query(parent))
708 else:
709 siblings = list(parent.children)
711 prev_item, next_item = None, None
712 try:
713 me = siblings.index(self)
714 except ValueError:
715 # Self not in parents.children or not in parents.pagination.
716 pass
717 else:
718 if me > 0:
719 prev_item = siblings[me - 1]
721 if me + 1 < len(siblings):
722 next_item = siblings[me + 1]
724 return prev_item, next_item
727class Attachment(Record):
728 """This represents a loaded attachment."""
730 is_attachment = True
732 def _is_hidden_by_parent_config(self) -> bool:
733 # Attachments are only considered hidden if they have been
734 # configured as such. If an explicit value for the _hidden field is not set,
735 # the value of the hidden option in the attachment configuration section of the
736 # parent's datamodel is checked. If that, too, is not set, attachments will be
737 # visible, even if their parent is hidden.
738 parent = self.parent
739 if parent is None:
740 return False
741 return parent.datamodel.attachment_config.hidden
743 @property
744 def record(self):
745 return self
747 @property
748 def attachment_filename(self):
749 return self.pad.db.to_fs_path(self._data["_path"])
751 @property
752 def parent(self):
753 """The associated record for this attachment."""
754 return self.pad.get(self._data["_attachment_for"], persist=self.pad.cache.is_persistent(self))
756 @cached_property
757 @deprecated(version="3.4.0", stacklevel=2)
758 def contents(self):
759 return FileContents(self.attachment_filename)
761 def get_fallback_record_label(self, lang):
762 return self["_id"]
764 def iter_source_filenames(self):
765 attachment_filename = self.attachment_filename
766 if self.alt != PRIMARY_ALT:
767 yield f"{attachment_filename}+{self.alt}.lr"
768 yield f"{attachment_filename}.lr"
769 yield attachment_filename
771 @property
772 def url_path(self):
773 # Attachments are only emitted for the primary alternative.
774 primary_alt = self.pad.config.primary_alternative or PRIMARY_ALT
775 return self._get_url_path(alt=primary_alt)
778class Image(Attachment):
779 """Specific class for image attachments."""
781 @cached_property
782 def _image_info(self):
783 return get_image_info(self.attachment_filename)
785 @cached_property
786 def exif(self):
787 """Provides access to the exif data."""
788 return read_exif(self.attachment_filename)
790 @property
791 def width(self):
792 """The width of the image if possible to determine."""
793 rv = self._image_info[1]
794 if rv is not None:
795 return rv
796 return Undefined("Width of image could not be determined.")
798 @property
799 def height(self):
800 """The height of the image if possible to determine."""
801 rv = self._image_info[2]
802 if rv is not None:
803 return rv
804 return Undefined("Height of image could not be determined.")
806 @property
807 def format(self):
808 """Returns the format of the image."""
809 rv = self._image_info[0]
810 if rv is not None:
811 return rv
812 return Undefined("The format of the image could not be determined.")
814 def thumbnail(self, width=None, height=None, mode=None, upscale=None, quality=None):
815 """Utility to create thumbnails."""
817 if mode is None:
818 mode = ThumbnailMode.DEFAULT
819 else:
820 mode = ThumbnailMode(mode)
822 if width is not None:
823 width = int(width)
824 if height is not None:
825 height = int(height)
827 return make_image_thumbnail(
828 _require_ctx(self),
829 self.attachment_filename,
830 self.url_path,
831 width=width,
832 height=height,
833 mode=mode,
834 upscale=upscale,
835 quality=quality,
836 )
839def require_ffmpeg(f):
840 """Decorator to help with error messages for ffmpeg template functions."""
841 # If both ffmpeg and ffprobe executables are available we don't need to
842 # override the function
843 if locate_executable("ffmpeg") and locate_executable("ffprobe"):
844 return f
846 @functools.wraps(f)
847 def wrapper(*args, **kwargs):
848 return Undefined("Unable to locate ffmpeg or ffprobe executable. Is it installed?")
850 return wrapper
853class Video(Attachment):
854 """Specific class for video attachments."""
856 @cached_property
857 def _video_info(self):
858 try:
859 return get_video_info(self.attachment_filename)
860 except RuntimeError:
861 return {}
863 @property
864 @require_ffmpeg
865 def width(self):
866 """Returns the width of the video if possible to determine."""
867 try:
868 return self._video_info["width"]
869 except KeyError:
870 return Undefined("The width of the video could not be determined.")
872 @property
873 @require_ffmpeg
874 def height(self):
875 """Returns the height of the video if possible to determine."""
876 try:
877 return self._video_info["height"]
878 except KeyError:
879 return Undefined("The height of the video could not be determined.")
881 @property
882 @require_ffmpeg
883 def duration(self):
884 """Returns the duration of the video if possible to determine."""
885 try:
886 return self._video_info["duration"]
887 except KeyError:
888 return Undefined("The duration of the video could not be determined.")
890 @require_ffmpeg
891 def frame(self, seek=None):
892 """Returns a VideoFrame object that is thumbnailable like an Image."""
893 duration = self.duration
894 if is_undefined(duration):
895 return Undefined("Unable to get video properties.")
897 if seek is None:
898 seek = duration / 2
899 return VideoFrame(self, seek)
902class VideoFrame:
903 """Representation of a specific frame in a VideoAttachment.
905 This is currently only useful for thumbnails, but in the future it might
906 work like an ImageAttachment.
907 """
909 def __init__(self, video, seek):
910 self.video = video
912 if not isinstance(seek, timedelta):
913 seek = timedelta(seconds=seek)
915 if seek < timedelta(0):
916 raise ValueError("Seek distance must not be negative")
917 if video.duration and seek > video.duration:
918 raise ValueError("Seek distance must not be outside the video duration")
920 self.seek = seek
922 def __str__(self):
923 raise NotImplementedError("It is currently not possible to use video frames directly, use .thumbnail().")
925 __unicode__ = __str__
927 @require_ffmpeg
928 def thumbnail(self, width=None, height=None, mode=None, upscale=None, quality=None):
929 """Utility to create thumbnails."""
930 if mode is None:
931 mode = ThumbnailMode.DEFAULT
932 else:
933 mode = ThumbnailMode(mode)
935 video = self.video
936 return make_video_thumbnail(
937 _require_ctx(video),
938 video.attachment_filename,
939 video.url_path,
940 seek=self.seek,
941 width=width,
942 height=height,
943 mode=mode,
944 upscale=upscale,
945 quality=quality,
946 )
949attachment_classes = {
950 "image": Image,
951 "video": Video,
952}
955class Query:
956 """Object that helps finding records. The default configuration
957 only finds pages.
958 """
960 def __init__(self, path, pad, alt=PRIMARY_ALT):
961 self.path = path
962 self.pad = pad
963 self.alt = alt
964 self._include_pages = True
965 self._include_attachments = False
966 self._order_by = None
967 self._filters = None
968 self._pristine = True
969 self._limit = None
970 self._offset = None
971 self._include_hidden = False
972 self._include_undiscoverable = False
973 self._page_num = None
974 self._filter_func = None
976 @property
977 def self(self):
978 """Returns the object this query starts out from."""
979 return self.pad.get(self.path, alt=self.alt)
981 def _clone(self, mark_dirty=False):
982 """Makes a flat copy but keeps the other data on it shared."""
983 rv = object.__new__(self.__class__)
984 rv.__dict__.update(self.__dict__)
985 if mark_dirty:
986 rv._pristine = False
987 return rv
989 def _get(self, id, persist=True, page_num=Ellipsis):
990 """Low level record access."""
991 if page_num is Ellipsis:
992 page_num = self._page_num
993 return self.pad.get(f"{self.path}/{id}", persist=persist, alt=self.alt, page_num=page_num)
995 def _matches(self, record):
996 if not self._include_hidden and record.is_hidden:
997 return False
998 if not self._include_undiscoverable and not record.is_discoverable:
999 return False
1000 for filter in self._filters or ():
1001 if not save_eval(filter, record):
1002 return False
1003 return True
1005 def _iterate(self):
1006 """Low level record iteration."""
1007 # If we iterate over children we also need to track those
1008 # dependencies. There are two ways in which we track them. The
1009 # first is through the start record of the query. If that does
1010 # not work for whatever reason (because it does not exist for
1011 # instance).
1012 self_record = self.pad.get(self.path, alt=self.alt)
1013 if self_record is not None:
1014 self.pad.db.track_record_dependency(self_record)
1016 # We also always want to record the path itself as dependency.
1017 ctx = get_ctx()
1018 if ctx is not None:
1019 ctx.record_dependency(self.pad.db.to_fs_path(self.path))
1021 for name, _, is_attachment in self.pad.db.iter_items(self.path, alt=self.alt):
1022 if not ((is_attachment == self._include_attachments) or (not is_attachment == self._include_pages)):
1023 continue
1025 record = self._get(name, persist=False)
1026 if self._matches(record):
1027 yield record
1029 def filter(self, expr):
1030 """Filters records by an expression."""
1031 rv = self._clone(mark_dirty=True)
1032 rv._filters = list(self._filters or ())
1033 if callable(expr):
1034 expr = _CallbackExpr(expr)
1035 rv._filters.append(expr)
1036 return rv
1038 def get_order_by(self):
1039 """Returns the order that should be used."""
1040 if self._order_by is not None:
1041 return self._order_by
1042 base_record = self.pad.get(self.path)
1043 if base_record is not None:
1044 if self._include_attachments and not self._include_pages:
1045 return base_record.datamodel.attachment_config.order_by
1046 if self._include_pages and not self._include_attachments:
1047 return base_record.datamodel.child_config.order_by
1048 # Otherwise the query includes either both or neither
1049 # attachments and/nor children. I have no idea which
1050 # value of order_by to use. We could punt and return
1051 # child_config.order_by, but for now, just return None.
1052 return None
1053 return None
1055 def include_hidden(self, value):
1056 """Controls whether hidden records should be included.
1058 By default, they are not.
1059 """
1060 rv = self._clone(mark_dirty=True)
1061 rv._include_hidden = value
1062 return rv
1064 def include_undiscoverable(self, value):
1065 """Controls whether undiscoverable records should be included as well."""
1066 rv = self._clone(mark_dirty=True)
1067 rv._include_undiscoverable = value
1068 return rv
1070 def request_page(self, page_num):
1071 """Requests a specific page number instead of the first."""
1072 rv = self._clone(mark_dirty=True)
1073 rv._page_num = page_num
1074 return rv
1076 def first(self):
1077 """Return the first matching record."""
1078 return next(iter(self), None)
1080 def all(self):
1081 """Loads all matching records as list."""
1082 return list(self)
1084 def order_by(self, *fields):
1085 """Sets the ordering of the query."""
1086 rv = self._clone()
1087 rv._order_by = fields or None
1088 return rv
1090 def offset(self, offset):
1091 """Sets the ordering of the query."""
1092 rv = self._clone(mark_dirty=True)
1093 rv._offset = offset
1094 return rv
1096 def limit(self, limit):
1097 """Sets the ordering of the query."""
1098 rv = self._clone(mark_dirty=True)
1099 rv._limit = limit
1100 return rv
1102 def count(self):
1103 """Counts all matched objects."""
1104 rv = 0
1105 for _ in self:
1106 rv += 1
1107 return rv
1109 def distinct(self, fieldname):
1110 """Set of unique values for the given field."""
1111 rv = set()
1113 for item in self:
1114 if fieldname in item._data:
1115 value = item._data[fieldname]
1116 if isinstance(value, (list, tuple)):
1117 rv |= set(value)
1118 elif not isinstance(value, Undefined):
1119 rv.add(value)
1121 return rv
1123 def get(self, id, page_num=Ellipsis):
1124 """Gets something by the local path."""
1125 # If we're not pristine, we need to query here
1126 if not self._pristine:
1127 q = self.filter(F._id == id)
1128 if page_num is not Ellipsis:
1129 q = q.request_page(page_num)
1130 return q.first()
1131 # otherwise we can load it directly.
1132 return self._get(id, page_num=page_num)
1134 def __bool__(self):
1135 return self.first() is not None
1137 __nonzero__ = __bool__
1139 def __iter__(self):
1140 """Iterates over all records matched."""
1141 iterable = self._iterate()
1143 order_by = self.get_order_by()
1144 if order_by:
1145 iterable = sorted(iterable, key=lambda x: x.get_sort_key(order_by))
1147 if self._offset is not None or self._limit is not None:
1148 iterable = islice(
1149 iterable,
1150 self._offset or 0,
1151 (self._offset or 0) + self._limit if self._limit else None,
1152 )
1154 yield from iterable
1156 def __repr__(self):
1157 alt_ = f" alt={self.alt!r}" if self.alt else ""
1158 return f"<{self.__class__.__name__} {self.path!r}{alt_}>"
1161class EmptyQuery(Query):
1162 def _get(self, id, persist=True, page_num=Ellipsis):
1163 pass
1165 def _iterate(self):
1166 """Low level record iteration."""
1167 return iter(())
1170class AttachmentsQuery(Query):
1171 """Specialized query class that only finds attachments."""
1173 def __init__(self, path, pad, alt=PRIMARY_ALT):
1174 Query.__init__(self, path, pad, alt=alt)
1175 self._include_pages = False
1176 self._include_attachments = True
1178 @property
1179 def images(self):
1180 """Filters to images."""
1181 return self.filter(F._attachment_type == "image")
1183 @property
1184 def videos(self):
1185 """Filters to videos."""
1186 return self.filter(F._attachment_type == "video")
1188 @property
1189 def audio(self):
1190 """Filters to audio."""
1191 return self.filter(F._attachment_type == "audio")
1193 @property
1194 def documents(self):
1195 """Filters to documents."""
1196 return self.filter(F._attachment_type == "document")
1198 @property
1199 def text(self):
1200 """Filters to plain text data."""
1201 return self.filter(F._attachment_type == "text")
1204def _iter_filename_choices(fn_base, alts, config, fallback=True):
1205 """Returns an iterator over all possible filename choices to .lr files
1206 below a base filename that matches any of the given alts.
1207 """
1208 # the order here is important as attachments can exist without a .lr
1209 # file and as such need to come second or the loading of raw data will
1210 # implicitly say the record exists.
1211 for alt in alts:
1212 if alt != PRIMARY_ALT and config.is_valid_alternative(alt):
1213 yield os.path.join(fn_base, f"contents+{alt}.lr"), alt, False
1215 if fallback or PRIMARY_ALT in alts:
1216 yield os.path.join(fn_base, "contents.lr"), PRIMARY_ALT, False
1218 for alt in alts:
1219 if alt != PRIMARY_ALT and config.is_valid_alternative(alt):
1220 yield f"{fn_base}+{alt}.lr", alt, True
1222 if fallback or PRIMARY_ALT in alts:
1223 yield f"{fn_base}.lr", PRIMARY_ALT, True
1226def _iter_content_files(dir_path, alts):
1227 """Returns an iterator over all existing content files below the given
1228 directory. This yields specific files for alts before it falls back
1229 to the primary alt.
1230 """
1231 for alt in alts:
1232 if alt == PRIMARY_ALT:
1233 continue
1234 if os.path.isfile(os.path.join(dir_path, f"contents+{alt}.lr")):
1235 yield alt
1236 if os.path.isfile(os.path.join(dir_path, "contents.lr")):
1237 yield PRIMARY_ALT
1240def _iter_datamodel_choices(datamodel_name, path, is_attachment=False):
1241 yield datamodel_name
1242 if not is_attachment:
1243 yield posixpath.basename(path).split(".")[0].replace("-", "_").lower()
1244 yield "page"
1245 yield "none"
1248def get_default_slug(record):
1249 """Compute the default slug for a page.
1251 This computes the default value of ``_slug`` for a page. The slug
1252 is computed by expanding the parent’s ``slug_format`` value.
1254 """
1255 parent = getattr(record, "parent", None)
1256 if parent is None:
1257 return ""
1258 return parent.datamodel.get_default_child_slug(record.pad, record)
1261default_slug_descriptor = property(get_default_slug)
1264class Database:
1265 def __init__(self, env, config=None):
1266 self.env = env
1267 if config is None:
1268 config = env.load_config()
1269 self.config = config
1270 self.datamodels = load_datamodels(env)
1271 self.flowblocks = load_flowblocks(env)
1273 def to_fs_path(self, path):
1274 """Convenience function to convert a path into an file system path."""
1275 return os.path.join(self.env.root_path, "content", untrusted_to_os_path(path))
1277 def load_raw_data(self, path, alt=PRIMARY_ALT, cls=None, fallback=True):
1278 """Internal helper that loads the raw record data. This performs
1279 very little data processing on the data.
1280 """
1281 path = cleanup_path(path)
1282 if cls is None:
1283 cls = dict
1285 fn_base = self.to_fs_path(path)
1287 rv = cls()
1288 rv_type = None
1290 choiceiter = _iter_filename_choices(fn_base, [alt], self.config, fallback=fallback)
1291 for fs_path, source_alt, is_attachment in choiceiter:
1292 # If we already determined what our return value is but the
1293 # type mismatches what we try now, we have to abort. Eg:
1294 # a page can not become an attachment or the other way round.
1295 if rv_type is not None and rv_type != is_attachment:
1296 break
1298 try:
1299 with open(fs_path, "rb") as f:
1300 if rv_type is None:
1301 rv_type = is_attachment
1302 for key, lines in metaformat.tokenize(f, encoding="utf-8"):
1303 if key not in rv:
1304 rv[key] = "".join(lines)
1305 except OSError as e:
1306 if e.errno not in (errno.ENOTDIR, errno.ENOENT, errno.EINVAL):
1307 raise
1308 if not is_attachment or not os.path.isfile(fs_path[:-3]):
1309 continue
1310 # Special case: we are loading an attachment but the meta
1311 # data file does not exist. In that case we still want to
1312 # record that we're loading an attachment.
1313 if is_attachment:
1314 rv_type = True
1316 if "_source_alt" not in rv:
1317 rv["_source_alt"] = source_alt
1319 if rv_type is None:
1320 return None
1322 rv["_path"] = path
1323 rv["_id"] = posixpath.basename(path)
1324 rv["_gid"] = hashlib.md5(path.encode("utf-8")).hexdigest()
1325 rv["_alt"] = alt
1326 if rv_type:
1327 rv["_attachment_for"] = posixpath.dirname(path)
1329 return rv
1331 def iter_items(self, path, alt=PRIMARY_ALT):
1332 """Iterates over all items below a path and yields them as
1333 tuples in the form ``(id, alt, is_attachment)``.
1334 """
1335 fn_base = self.to_fs_path(path)
1337 if alt is None:
1338 alts = self.config.list_alternatives()
1339 single_alt = False
1340 else:
1341 alts = [alt]
1342 single_alt = True
1344 choiceiter = _iter_filename_choices(fn_base, alts, self.config)
1346 for fs_path, _actual_alt, is_attachment in choiceiter:
1347 if not os.path.isfile(fs_path):
1348 continue
1350 # This path is actually for an attachment, which means that we
1351 # cannot have any items below it and will just abort with an
1352 # empty iterator.
1353 if is_attachment:
1354 break
1356 try:
1357 dir_path = os.path.dirname(fs_path)
1358 for filename in os.listdir(dir_path):
1359 if not isinstance(filename, str):
1360 try:
1361 filename = filename.decode(fs_enc)
1362 except UnicodeError:
1363 continue
1365 if filename.endswith(".lr") or self.env.is_uninteresting_source_name(filename):
1366 continue
1368 # We found an attachment. Attachments always live
1369 # below the primary alt, so we report it as such.
1370 if os.path.isfile(os.path.join(dir_path, filename)):
1371 yield filename, PRIMARY_ALT, True
1373 # We found a directory, let's make sure it contains a
1374 # contents.lr file (or a contents+alt.lr file).
1375 else:
1376 for content_alt in _iter_content_files(os.path.join(dir_path, filename), alts):
1377 yield filename, content_alt, False
1378 # If we want a single alt, we break here so
1379 # that we only produce a single result.
1380 # Otherwise this would also return the primary
1381 # fallback here.
1382 if single_alt:
1383 break
1384 except OSError as e:
1385 if e.errno != errno.ENOENT:
1386 raise
1387 continue
1389 # If we reach this point, we found our parent, so we can stop
1390 # searching for more at this point.
1391 break
1393 def get_datamodel_for_raw_data(self, raw_data, pad=None):
1394 """Returns the datamodel that should be used for a specific raw
1395 data. This might require the discovery of a parent object through
1396 the pad.
1397 """
1398 path = raw_data["_path"]
1399 is_attachment = bool(raw_data.get("_attachment_for"))
1400 datamodel = (raw_data.get("_model") or "").strip() or None
1401 return self.get_implied_datamodel(path, is_attachment, pad, datamodel=datamodel)
1403 def iter_dependent_models(self, datamodel):
1404 seen = set()
1406 def deep_find(datamodel):
1407 seen.add(datamodel)
1409 if datamodel.parent is not None and datamodel.parent not in seen:
1410 deep_find(datamodel.parent)
1412 for related_dm_name in (
1413 datamodel.child_config.model,
1414 datamodel.attachment_config.model,
1415 ):
1416 dm = self.datamodels.get(related_dm_name)
1417 if dm is not None and dm not in seen:
1418 deep_find(dm)
1420 deep_find(datamodel)
1421 seen.discard(datamodel)
1422 return iter(seen)
1424 def get_implied_datamodel(self, path, is_attachment=False, pad=None, datamodel=None):
1425 """Looks up a datamodel based on the information about the parent
1426 of a model.
1427 """
1428 model = datamodel
1430 # Only look for a datamodel if there was not defined.
1431 if model is None:
1432 parent = posixpath.dirname(path)
1434 # If we hit the root, and there is no model defined we need
1435 # to make sure we do not recurse onto ourselves.
1436 if parent != path:
1437 if pad is None:
1438 pad = self.new_pad()
1439 parent_obj = pad.get(parent)
1440 if parent_obj is not None:
1441 if is_attachment:
1442 model = parent_obj.datamodel.attachment_config.model
1443 else:
1444 model = parent_obj.datamodel.child_config.model
1446 for dm_name in _iter_datamodel_choices(model, path, is_attachment):
1447 # If that datamodel exists, let's roll with it.
1448 datamodel = self.datamodels.get(dm_name)
1449 if datamodel is not None:
1450 return datamodel
1452 raise AssertionError("Did not find an appropriate datamodel. That should never happen.")
1454 def get_attachment_type(self, path):
1455 """Gets the attachment type for a path."""
1456 return self.config["ATTACHMENT_TYPES"].get(posixpath.splitext(path)[1].lower())
1458 def track_record_dependency(self, record):
1459 ctx = get_ctx()
1460 if ctx is not None:
1461 for filename in record.iter_source_filenames():
1462 if isinstance(record, Attachment):
1463 # For Attachments, the actually attachment data
1464 # does not affect the URL of the attachment.
1465 affects_url = filename != record.attachment_filename
1466 else:
1467 affects_url = True
1468 ctx.record_dependency(filename, affects_url=affects_url)
1469 if isinstance(record, VirtualSourceObject):
1470 ctx.record_virtual_dependency(record)
1471 if getattr(record, "datamodel", None) and record.datamodel.filename:
1472 ctx.record_dependency(record.datamodel.filename)
1473 for dep_model in self.iter_dependent_models(record.datamodel):
1474 if dep_model.filename:
1475 ctx.record_dependency(dep_model.filename)
1476 # XXX: In the case that our datamodel is implied, then the
1477 # datamodel depends on the datamodel(s) of our parent(s).
1478 # We do not currently record that.
1479 return record
1481 def process_data(self, data, datamodel, pad):
1482 # Automatically fill in slugs
1483 if not data["_slug"]:
1484 data["_slug"] = default_slug_descriptor
1485 else:
1486 data["_slug"] = data["_slug"].strip("/")
1488 # For attachments figure out the default attachment type if it's
1489 # not yet provided.
1490 if is_undefined(data["_attachment_type"]) and data["_attachment_for"]:
1491 data["_attachment_type"] = self.get_attachment_type(data["_path"])
1493 # Automatically fill in templates
1494 if is_undefined(data["_template"]):
1495 data["_template"] = datamodel.get_default_template_name()
1497 @staticmethod
1498 def get_record_class(datamodel, raw_data):
1499 """Returns the appropriate record class for a datamodel and raw data."""
1500 is_attachment = bool(raw_data.get("_attachment_for"))
1501 if not is_attachment:
1502 return Page
1503 attachment_type = raw_data["_attachment_type"]
1504 return attachment_classes.get(attachment_type, Attachment)
1506 def new_pad(self):
1507 """Creates a new pad object for this database."""
1508 return Pad(self)
1511def _split_alt_from_url(config, clean_path):
1512 primary = config.primary_alternative
1514 # The alternative system is not configured, just return
1515 if primary is None:
1516 return None, clean_path
1518 # First try to find alternatives that are identified by a prefix.
1519 for prefix, alt in config.get_alternative_url_prefixes():
1520 if clean_path.startswith(prefix):
1521 return alt, clean_path[len(prefix) :].strip("/")
1522 # Special case which is the URL root.
1523 if prefix.strip("/") == clean_path:
1524 return alt, ""
1526 # Now find alternatives that are identified by a suffix.
1527 for suffix, alt in config.get_alternative_url_suffixes():
1528 if clean_path.endswith(suffix):
1529 return alt, clean_path[: -len(suffix)].strip("/")
1531 # If we have a primary alternative without a prefix and suffix, we can
1532 # return that one.
1533 if config.primary_alternative_is_rooted:
1534 return None, clean_path
1536 return None, None
1539class Pad:
1540 def __init__(self, db: Database):
1541 self.db = db
1542 self.cache = RecordCache(db.config["EPHEMERAL_RECORD_CACHE_SIZE"])
1543 self.databags = Databags(db.env)
1545 @property
1546 def config(self) -> Config:
1547 """The config for this pad."""
1548 return self.db.config
1550 @property
1551 def env(self) -> Environment:
1552 """The env for this pad."""
1553 return self.db.env
1555 @deprecated("use Pad.make_url instead", version="3.4.0")
1556 def make_absolute_url(self, url):
1557 """Given a URL this makes it absolute if this is possible."""
1558 base_url = self.db.config["PROJECT"].get("url")
1559 if base_url is None:
1560 raise RuntimeError("To use absolute URLs you need to configure the URL in the project config.")
1561 return urljoin(base_url.rstrip("/") + "/", url.lstrip("/"))
1563 def make_url(self, url, base_url=None, absolute=None, external=None):
1564 """Helper method that creates a finalized URL based on the parameters
1565 provided and the config.
1567 :param url: URL path (starting with "/") relative to the
1568 configured base_path.
1570 :param base_url: Base URL path (starting with "/") relative to
1571 the configured base_path.
1573 """
1574 url_style = self.db.config.url_style
1575 if absolute is None:
1576 absolute = url_style == "absolute"
1577 if external is None:
1578 external = url_style == "external"
1579 if external:
1580 external_base_url = self.db.config.base_url
1581 if external_base_url is None:
1582 raise RuntimeError("To use absolute URLs you need to configure the URL in the project config.")
1583 return urljoin(external_base_url, url.lstrip("/"))
1584 if absolute:
1585 return urljoin(self.db.config.base_path, url.lstrip("/"))
1586 if base_url is None:
1587 raise RuntimeError("Cannot calculate a relative URL if no base URL has been provided.")
1588 return make_relative_url(base_url, url)
1590 def resolve_url_path(self, url_path, include_invisible=False, include_assets=True, alt_fallback=True):
1591 """Given a URL path this will find the correct record which also
1592 might be an attachment. If a record cannot be found or is unexposed
1593 the return value will be `None`.
1594 """
1595 try:
1596 clean_path = cleanup_url_path(url_path).strip("/")
1597 except ValueError:
1598 return None
1600 # Split off the alt and if no alt was found, point it to the
1601 # primary alternative. If the clean path comes back as `None`
1602 # then the config does not include a rooted alternative and we
1603 # have to skip handling of regular records.
1604 alt, clean_path = _split_alt_from_url(self.db.config, clean_path)
1605 if clean_path is not None:
1606 if not alt:
1607 if alt_fallback:
1608 alt = self.db.config.primary_alternative or PRIMARY_ALT
1609 else:
1610 alt = PRIMARY_ALT
1611 node = self.get_root(alt=alt)
1612 if node is None:
1613 raise RuntimeError("Tree root could not be found.")
1615 pieces = clean_path.split("/")
1616 if pieces == [""]:
1617 pieces = []
1619 rv = node.resolve_url_path(pieces)
1620 if rv is not None and (include_invisible or rv.is_visible):
1621 return rv
1623 if include_assets:
1624 return self.asset_root.resolve_url_path(pieces)
1625 return None
1627 def get_root(self, alt=None):
1628 """The root page of the database."""
1629 if alt is None:
1630 alt = self.config.primary_alternative or PRIMARY_ALT
1631 return self.get("/", alt=alt, persist=True)
1633 root = property(get_root)
1635 @cached_property
1636 def asset_root(self):
1637 """The root of the asset tree.
1639 This root represents the logical merging of any theme asset trees with the
1640 main project asset tree.
1641 """
1642 env = self.env
1643 asset_paths = (Path(root, "assets") for root in chain([env.root_path], env.theme_paths))
1644 return get_asset_root(self, asset_paths)
1646 @property
1647 @deprecated(version="3.4.0", stacklevel=2)
1648 def theme_asset_roots(self):
1649 """The root of the asset tree of each theme.
1651 As of Lektor 3.4.0, asset trees from any active themes are logically merged into
1652 a single tree, accessible via Pad.asset_root. Accordingly, `theme_asset_roots`
1653 alway returns an empty list.
1655 """
1656 return []
1658 def get_all_roots(self):
1659 """Returns all the roots for building."""
1660 rv = [self.get_root(alt=alt) for alt in self.db.config.list_alternatives()]
1661 # If we don't have any alternatives, then we go with the implied
1662 # root.
1663 if not rv and self.root:
1664 rv = [self.root]
1666 rv.append(self.asset_root)
1667 return rv
1669 def get_virtual(self, record, virtual_path):
1670 """Resolves a virtual path below a record."""
1671 pieces = virtual_path.strip("/").split("/")
1672 if not pieces or pieces == [""]:
1673 return record
1675 if pieces[0].isdigit():
1676 if len(pieces) == 1:
1677 return self.get(record._data["_path"], alt=record.alt, page_num=int(pieces[0]))
1678 return None
1680 resolver = self.env.virtual_sources.get(pieces[0])
1681 if resolver is None:
1682 return None
1684 return resolver(record, pieces[1:])
1686 def get(self, path, alt=None, page_num=None, persist=True, allow_virtual=True):
1687 """Loads a record by path."""
1688 if alt is None:
1689 alt = self.config.primary_alternative or PRIMARY_ALT
1690 virt_markers = path.count("@")
1692 # If the virtual marker is included, we also want to look up the
1693 # virtual path below an item. Special case: if virtual paths are
1694 # not allowed but one was passed, we just return `None`.
1695 if virt_markers == 1:
1696 if page_num is not None:
1697 raise RuntimeError(
1698 "Cannot use both virtual paths and explicit page number lookups. You need to one or the other."
1699 )
1700 if not allow_virtual:
1701 return None
1702 path, virtual_path = path.split("@", 1)
1703 rv = self.get(path, alt=alt, page_num=page_num, persist=persist)
1704 if rv is None:
1705 return None
1706 return self.get_virtual(rv, virtual_path)
1708 # Sanity check: there must only be one or things will get weird.
1709 if virt_markers > 1:
1710 return None
1712 path = cleanup_path(path)
1713 virtual_path = None
1714 if page_num is not None:
1715 virtual_path = str(page_num)
1717 rv = self.cache.get(path, alt, virtual_path)
1718 if rv is not Ellipsis:
1719 if rv is not None:
1720 self.db.track_record_dependency(rv)
1721 return rv
1723 raw_data = self.db.load_raw_data(path, alt=alt)
1724 if raw_data is None:
1725 self.cache.remember_as_missing(path, alt, virtual_path)
1726 return None
1728 rv = self.instance_from_data(raw_data, page_num=page_num)
1730 if persist:
1731 self.cache.persist(rv)
1732 else:
1733 self.cache.remember(rv)
1735 return self.db.track_record_dependency(rv)
1737 def alt_exists(self, path, alt=PRIMARY_ALT, fallback=False):
1738 """Checks if an alt exists."""
1739 path = cleanup_path(path)
1740 if "@" in path:
1741 return False
1743 # If we find the path in the cache, check if it was loaded from
1744 # the right source alt.
1745 rv = self.get(path, alt)
1746 if rv is not None:
1747 if rv["_source_alt"] == alt:
1748 return True
1749 if fallback or (rv["_source_alt"] == PRIMARY_ALT and alt == self.config.primary_alternative):
1750 return True
1751 return False
1753 return False
1755 def get_asset(self, path):
1756 """Loads an asset by path."""
1757 clean_path = cleanup_path(path).strip("/")
1759 asset = self.asset_root
1760 if clean_path:
1761 for piece in clean_path.split("/"):
1762 asset = asset.get_child(piece)
1763 if asset is None:
1764 return None
1765 return asset
1767 def instance_from_data(self, raw_data, datamodel=None, page_num=None):
1768 """This creates an instance from the given raw data."""
1769 if datamodel is None:
1770 datamodel = self.db.get_datamodel_for_raw_data(raw_data, self)
1771 data = datamodel.process_raw_data(raw_data, self)
1772 self.db.process_data(data, datamodel, self)
1773 cls = self.db.get_record_class(datamodel, data)
1774 return cls(self, data, page_num=page_num)
1776 def query(self, path=None, alt=PRIMARY_ALT):
1777 """Queries the database either at root level or below a certain
1778 path. This is the recommended way to interact with toplevel data.
1779 The alternative is to work with the :attr:`root` document.
1780 """
1781 # Don't accidentally pass `None` down to the query as this might
1782 # do some unexpected things.
1783 if alt is None:
1784 alt = PRIMARY_ALT
1785 return Query(path="/" + (path or "").strip("/"), pad=self, alt=alt).include_hidden(True)
1788class TreeItem:
1789 """Represents a single tree item and all the alts within it."""
1791 def __init__(self, tree, path, alts, primary_record):
1792 self.tree = tree
1793 self.path = path
1794 self.alts = alts
1795 self._primary_record = primary_record
1797 @property
1798 def id(self):
1799 """The local ID of the item."""
1800 return posixpath.basename(self.path)
1802 @property
1803 def exists(self):
1804 """True iff metadata exists for the item.
1806 If metadata exists for any alt, including the fallback (PRIMARY_ALT).
1808 Note that for attachments without metadata, this is currently False.
1809 But note that if is_attachment is True, the attachment file does exist.
1810 """
1811 # FIXME: this should probably be changed to return True for attachments,
1812 # even those without metadata.
1813 return self._primary_record is not None
1815 @property
1816 def is_attachment(self):
1817 """True iff item is an attachment."""
1818 if self._primary_record is None:
1819 return False
1820 return self._primary_record.is_attachment
1822 @property
1823 def is_visible(self):
1824 """True iff item is not hidden."""
1825 # XXX: This is send out from /api/recordinfo but appears to be
1826 # unused by the React app
1827 if self._primary_record is None:
1828 return True
1829 return self._primary_record.is_visible
1831 @property
1832 def can_be_deleted(self):
1833 """True iff item can be deleted."""
1834 if self.path == "/" or not self.exists:
1835 return False
1836 return self.is_attachment or not self._datamodel.protected
1838 @property
1839 def _datamodel(self):
1840 if self._primary_record is None:
1841 return None
1842 return self._primary_record.datamodel
1844 def get_record_label_i18n(self, alt=PRIMARY_ALT):
1845 """Get record label translations for specific alt."""
1846 record = self.alts[alt].record
1847 if record is None:
1848 # generate a reasonable fallback
1849 # ("en" is the magical fallback lang)
1850 label = self.id.replace("-", " ").replace("_", " ").title()
1851 return {"en": label or "(Index)"}
1852 return record.get_record_label_i18n()
1854 @property
1855 def can_have_children(self):
1856 """True iff the item can contain subpages."""
1857 if self._primary_record is None or self.is_attachment:
1858 return False
1859 return self._datamodel.has_own_children
1861 @property
1862 def implied_child_datamodel(self):
1863 """The name of the default datamodel for children of this page, if any."""
1864 datamodel = self._datamodel
1865 return datamodel.child_config.model if datamodel else None
1867 @property
1868 def can_have_attachments(self):
1869 """True iff the item can contain attachments."""
1870 if self._primary_record is None or self.is_attachment:
1871 return False
1872 return self._datamodel.has_own_attachments
1874 @property
1875 def attachment_type(self):
1876 """The type of an attachment.
1878 E.g. "image", "video", or None if type is unknown.
1879 """
1880 if self._primary_record is None or not self.is_attachment:
1881 return None
1882 return self._primary_record["_attachment_type"] or None
1884 def get_parent(self):
1885 """Returns the parent item."""
1886 if self.path == "/":
1887 return None
1888 return self.tree.get(posixpath.dirname(self.path))
1890 def get(self, path):
1891 """Returns a child within this item."""
1892 # XXX: Unused?
1893 return self.tree.get(posixpath.join(self.path, path))
1895 def _get_child_ids(self, include_attachments=True, include_pages=True):
1896 """Returns a sorted list of just the IDs of existing children."""
1897 db = self.tree.pad.db
1898 keep_attachments = include_attachments and self.can_have_attachments
1899 keep_pages = include_pages and self.can_have_children
1900 names = {
1901 name
1902 for name, _, is_attachment in db.iter_items(self.path, alt=None)
1903 if (keep_attachments if is_attachment else keep_pages)
1904 }
1905 return sorted(names, key=lambda name: name.lower())
1907 def iter_children(self, include_attachments=True, include_pages=True, order_by=None):
1908 """Iterates over all children"""
1909 children = (
1910 self.tree.get(posixpath.join(self.path, name), persist=False)
1911 for name in self._get_child_ids(include_attachments, include_pages)
1912 )
1913 if order_by is not None:
1914 children = sorted(children, key=methodcaller("get_sort_key", order_by))
1915 return children
1917 def get_children(
1918 self,
1919 offset=0,
1920 limit=None,
1921 include_attachments=True,
1922 include_pages=True,
1923 order_by=None,
1924 ):
1925 """Returns a slice of children."""
1926 # XXX: this method appears unused?
1927 end = None
1928 if limit is not None:
1929 end = offset + limit
1930 children = self.iter_children(include_attachments, include_pages, order_by)
1931 return list(islice(children, offset, end))
1933 def iter_attachments(self, order_by=None):
1934 """Return an iterable of this records attachments.
1936 By default, the attachments are sorted as specified by
1937 ``[attachments]order_by`` in this records datamodel.
1938 """
1939 if order_by is None:
1940 dm = self._datamodel
1941 if dm is not None:
1942 order_by = dm.attachment_config.order_by
1943 return self.iter_children(include_pages=False, order_by=order_by)
1945 def iter_subpages(self, order_by=None):
1946 """Return an iterable of this records sub-pages.
1948 By default, the records are sorted as specified by
1949 ``[children]order_by`` in this records datamodel.
1951 NB: This method should probably be called ``iter_children``,
1952 but that name was already taken.
1953 """
1954 if order_by is None:
1955 dm = self._datamodel
1956 if dm is not None:
1957 order_by = dm.child_config.order_by
1958 return self.iter_children(include_attachments=False, order_by=order_by)
1960 def get_sort_key(self, order_by):
1961 if self._primary_record is None:
1963 def sort_key(fieldspec):
1964 if fieldspec.startswith("-"):
1965 field, reverse = fieldspec[1:], True
1966 else:
1967 field, reverse = fieldspec.lstrip("+"), False
1968 value = self.id if field == "_id" else None
1969 return _CmpHelper(value, reverse)
1971 return [sort_key(fieldspec) for fieldspec in order_by]
1972 return self._primary_record.get_sort_key(order_by)
1974 def __repr__(self):
1975 return "<TreeItem {!r}{}>".format(
1976 self.path,
1977 self.is_attachment and " attachment" or "",
1978 )
1981class Alt:
1982 def __init__(self, id, record, is_primary_overlay, name_i18n):
1983 self.id = id
1984 self.record = record
1985 self.is_primary_overlay = is_primary_overlay
1986 self.name_i18n = name_i18n
1987 self.exists = record is not None and os.path.isfile(record.source_filename)
1989 def __repr__(self):
1990 return "<Alt {!r}{}>".format(self.id, self.exists and "*" or "")
1993class Tree:
1994 """Special object that can be used to get a broader insight into the
1995 database in a way that is not bound to the alt system directly.
1997 This wraps a pad and provides additional ways to interact with the data
1998 of the database in a way that is similar to how the data is actually laid
1999 out on the file system and not as the data is represented. Primarily the
2000 difference is how alts are handled. Where the pad resolves the alts
2001 automatically to make the handling automatic, the tree will give access
2002 to the underlying alts automatically.
2003 """
2005 def __init__(self, pad):
2006 # Note that in theory different alts can disagree on what
2007 # datamodel they use but this is something that is really not
2008 # supported. This cannot happen if you edit based on the admin
2009 # panel and if you edit it manually and screw up that part, we
2010 # cannot really do anything about it.
2011 #
2012 # We will favor the datamodel from the fallback record
2013 # (alt=PRIMARY_ALT) if it exists. If it does not, we will
2014 # use the data for the primary alternative. If data does not exist
2015 # for either of those, we will pick from among the other
2016 # existing alts quasi-randomly.
2017 #
2018 # Here we construct a list of configured alts in preference order
2019 config = pad.db.config
2020 alt_info = OrderedDict()
2021 if config.primary_alternative:
2022 # Alternatives are configured
2023 alts = [PRIMARY_ALT]
2024 alts.append(config.primary_alternative)
2025 alts.extend(alt for alt in config.list_alternatives() if alt != config.primary_alternative)
2026 for alt in alts:
2027 alt_info[alt] = {
2028 "is_primary_overlay": alt == config.primary_alternative,
2029 "name_i18n": config.get_alternative(alt)["name"],
2030 }
2031 else:
2032 alt_info[PRIMARY_ALT] = {
2033 "is_primary_overlay": True,
2034 "name_i18n": {"en": "Primary"},
2035 }
2037 self.pad = pad
2038 self._alt_info = alt_info
2040 def get(self, path=None, persist=True):
2041 """Returns a path item at the given node."""
2042 path = "/" + (path or "").strip("/")
2043 alts = {}
2044 primary_record = None
2045 for alt, alt_info in self._alt_info.items():
2046 record = self.pad.get(path, alt=alt, persist=persist, allow_virtual=False)
2047 if primary_record is None:
2048 primary_record = record
2049 alts[alt] = Alt(alt, record, **alt_info)
2050 return TreeItem(self, path, alts, primary_record)
2052 def iter_children(self, path=None, include_attachments=True, include_pages=True, order_by=None):
2053 """Iterates over all children below a path"""
2054 # XXX: this method is unused?
2055 path = "/" + (path or "").strip("/")
2056 return self.get(path, persist=False).iter_children(include_attachments, include_pages, order_by)
2058 def get_children(
2059 self,
2060 path=None,
2061 offset=0,
2062 limit=None,
2063 include_attachments=True,
2064 include_pages=True,
2065 order_by=None,
2066 ):
2067 """Returns a slice of children."""
2068 # XXX: this method is unused?
2069 path = "/" + (path or "").strip("/")
2070 return self.get(path, persist=False).get_children(offset, limit, include_attachments, include_pages, order_by)
2072 def edit(self, path, is_attachment=None, alt=PRIMARY_ALT, datamodel=None):
2073 """Edits a record by path."""
2074 return make_editor_session(
2075 self.pad,
2076 cleanup_path(path),
2077 alt=alt,
2078 is_attachment=is_attachment,
2079 datamodel=datamodel,
2080 )
2083class RecordCache:
2084 """The record cache holds records either in an persistent or ephemeral
2085 section which helps the pad not load records it already saw.
2086 """
2088 def __init__(self, ephemeral_cache_size=1000):
2089 self.persistent = {}
2090 self.ephemeral = LRUCache(ephemeral_cache_size)
2092 @staticmethod
2093 def _get_cache_key(record_or_path, alt=PRIMARY_ALT, virtual_path=None):
2094 if isinstance(record_or_path, str):
2095 path = record_or_path.strip("/")
2096 else:
2097 path, virtual_path = split_virtual_path(record_or_path.path)
2098 path = path.strip("/")
2099 virtual_path = virtual_path or None
2100 alt = record_or_path.alt
2101 return (path, alt, virtual_path)
2103 def flush(self):
2104 """Flushes the cache"""
2105 self.persistent.clear()
2106 self.ephemeral.clear()
2108 def is_persistent(self, record):
2109 """Indicates if a record is in the persistent record cache."""
2110 cache_key = self._get_cache_key(record)
2111 return cache_key in self.persistent
2113 def remember(self, record):
2114 """Remembers the record in the record cache."""
2115 cache_key = self._get_cache_key(record)
2116 if cache_key not in self.persistent and cache_key not in self.ephemeral:
2117 self.ephemeral[cache_key] = record
2119 def persist(self, record):
2120 """Persists a record. This will put it into the persistent cache."""
2121 cache_key = self._get_cache_key(record)
2122 self.persistent[cache_key] = record
2123 try:
2124 del self.ephemeral[cache_key]
2125 except KeyError:
2126 pass
2128 def persist_if_cached(self, record):
2129 """If the record is already ephemerally cached, this promotes it to
2130 the persistent cache section.
2131 """
2132 cache_key = self._get_cache_key(record)
2133 if cache_key in self.ephemeral:
2134 self.persist(record)
2136 def get(self, path, alt=PRIMARY_ALT, virtual_path=None):
2137 """Looks up a record from the cache."""
2138 cache_key = self._get_cache_key(path, alt, virtual_path)
2139 rv = self.persistent.get(cache_key, Ellipsis)
2140 if rv is not Ellipsis:
2141 return rv
2142 rv = self.ephemeral.get(cache_key, Ellipsis)
2143 if rv is not Ellipsis:
2144 return rv
2145 return Ellipsis
2147 def remember_as_missing(self, path, alt=PRIMARY_ALT, virtual_path=None):
2148 cache_key = self._get_cache_key(path, alt, virtual_path)
2149 self.persistent.pop(cache_key, None)
2150 self.ephemeral[cache_key] = None