Coverage for src/lektor_ng/build_programs.py: 97%
137 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-03 22:08 +0000
1import os
2import shutil
3from itertools import chain
5from lektor_ng.assets import Directory, File
6from lektor_ng.constants import PRIMARY_ALT
7from lektor_ng.db import Attachment, Page
8from lektor_ng.exception import LektorException
11class BuildError(LektorException):
12 pass
15builtin_build_programs = []
18def buildprogram(source_cls):
19 def decorator(builder_cls):
20 builtin_build_programs.append((source_cls, builder_cls))
21 return builder_cls
23 return decorator
26class SourceInfo:
27 """Holds some information about a source file for indexing into the
28 build state.
29 """
31 def __init__(self, path, filename, alt=PRIMARY_ALT, type="unknown", title_i18n=None):
32 self.path = path
33 self.alt = alt
34 self.filename = filename
35 self.type = type
36 self.title_i18n = {}
38 en_title = self.path
39 if "en" in title_i18n:
40 en_title = title_i18n["en"]
41 for key, value in title_i18n.items():
42 if key == "en":
43 continue
44 if value != en_title:
45 self.title_i18n[key] = value
46 self.title_i18n["en"] = en_title
49class BuildProgram:
50 def __init__(self, source, build_state):
51 self.source = source
52 self.build_state = build_state
53 self.artifacts = []
54 self._built = False
56 @property
57 def primary_artifact(self):
58 """Returns the primary artifact for this build program. By
59 default this is the first artifact produced. This needs to be the
60 one that corresponds to the URL of the source if it has one.
61 """
62 try:
63 return self.artifacts[0]
64 except IndexError:
65 return None
67 def describe_source_record(self):
68 """Can be used to describe the source info by returning a
69 :class:`SourceInfo` object. This is indexed by the builder into
70 the build state so that the UI can quickly find files without
71 having to scan the file system.
72 """
74 def build(self):
75 """Invokes the build program."""
76 if self._built:
77 raise RuntimeError("This build program was already used.")
78 self._built = True
80 self.produce_artifacts()
82 sub_artifacts = []
83 failures = []
85 gen = self.build_state.builder
87 def _build(artifact, build_func):
88 ctx = gen.build_artifact(artifact, build_func)
89 if ctx is not None:
90 if ctx.exc_info is not None:
91 failures.append(ctx.exc_info)
92 else:
93 sub_artifacts.extend(ctx.sub_artifacts)
95 # Step one is building the artifacts that this build program
96 # knows about.
97 for artifact in self.artifacts:
98 _build(artifact, self.build_artifact)
100 # For as long as our ctx keeps producing sub artifacts, we
101 # want to process them as well.
102 while sub_artifacts and not failures:
103 artifact, build_func = sub_artifacts.pop()
104 _build(artifact, build_func)
106 # If we failed anywhere we want to mark *all* artifacts as dirty.
107 # This means that if a sub-artifact fails we also rebuild the
108 # parent next time around.
109 if failures:
110 for artifact in self.artifacts:
111 artifact.set_dirty_flag()
113 def produce_artifacts(self):
114 """This produces the artifacts for building. Usually this only
115 produces a single artifact.
116 """
118 def declare_artifact(self, artifact_name, sources=None, extra=None):
119 """This declares an artifact to be built in this program."""
120 self.artifacts.append(
121 self.build_state.new_artifact(
122 artifact_name=artifact_name,
123 sources=sources,
124 source_obj=self.source,
125 extra=extra,
126 )
127 )
129 def build_artifact(self, artifact):
130 """This is invoked for each artifact declared."""
132 def iter_child_sources(self):
133 """This allows a build program to produce children that also need
134 building. An individual build never recurses down to this, but
135 a `build_all` will use this.
136 """
137 # pylint: disable=no-self-use
138 return iter(())
141@buildprogram(Page)
142class PageBuildProgram(BuildProgram):
143 def describe_source_record(self):
144 # When we describe the source record we need to consider that a
145 # page has multiple source file names but only one will actually
146 # be used. The order of the source iter is in order the files are
147 # attempted to be read. So we go with the first that actually
148 # exists and then return that.
149 for filename in self.source.iter_source_filenames():
150 if os.path.isfile(filename):
151 return SourceInfo(
152 path=self.source.path,
153 alt=self.source["_source_alt"],
154 filename=filename,
155 type="page",
156 title_i18n=self.source.get_record_label_i18n(),
157 )
158 return None
160 def produce_artifacts(self):
161 pagination_enabled = self.source.datamodel.pagination_config.enabled
163 if self.source.is_visible and (self.source.page_num is not None or not pagination_enabled):
164 artifact_name = self.source.url_path
165 if artifact_name.endswith("/"):
166 artifact_name += "index.html"
168 self.declare_artifact(artifact_name, sources=list(self.source.iter_source_filenames()))
170 def build_artifact(self, artifact):
171 # Record dependecies on all our sources and datamodel
172 self.source.pad.db.track_record_dependency(self.source)
174 try:
175 self.source.url_path.encode("ascii")
176 except UnicodeError as error:
177 raise BuildError(
178 "The URL for this record contains non ASCII "
179 "characters. This is currently not supported "
180 f"for portability reasons ({self.source.url_path!r})."
181 ) from error
183 artifact.render_template_into(self.source["_template"], this=self.source)
185 def _iter_paginated_children(self):
186 total = self.source.datamodel.pagination_config.count_pages(self.source)
187 for page_num in range(1, total + 1):
188 yield Page(self.source.pad, self.source._data, page_num=page_num)
190 def iter_child_sources(self):
191 p_config = self.source.datamodel.pagination_config
192 pagination_enabled = p_config.enabled
193 child_sources = []
195 # So this requires a bit of explanation:
196 #
197 # the basic logic is that if we have pagination enabled then we
198 # need to consider two cases:
199 #
200 # 1. our build program has page_num = None which means that we
201 # are not yet pointing to a page. In that case we want to
202 # iter over all children which will yield the pages.
203 # 2. we are pointing to a page, then our child sources are the
204 # items that are shown on that page.
205 #
206 # In addition, attachments and pages excluded from pagination are
207 # linked to the page with page_num = None.
208 #
209 # If pagination is disabled, all children and attachments are linked
210 # to this page.
211 all_children = self.source.children.include_undiscoverable(True)
212 all_children = all_children.include_hidden(True)
213 if pagination_enabled:
214 if self.source.page_num is None:
215 child_sources.append(self._iter_paginated_children())
216 pq = p_config.get_pagination_query(self.source)
217 child_sources.append(set(all_children) - set(pq))
218 child_sources.append(self.source.attachments)
219 else:
220 child_sources.append(self.source.pagination.items)
221 else:
222 child_sources.append(all_children)
223 child_sources.append(self.source.attachments)
225 return chain(*child_sources)
228@buildprogram(Attachment)
229class AttachmentBuildProgram(BuildProgram):
230 def describe_source_record(self):
231 return SourceInfo(
232 path=self.source.path,
233 alt=self.source.alt,
234 filename=self.source.attachment_filename,
235 type="attachment",
236 title_i18n={"en": self.source["_id"]},
237 )
239 def produce_artifacts(self):
240 primary_alt = self.build_state.config.primary_alternative or PRIMARY_ALT
241 if self.source.is_visible and self.source.alt == primary_alt:
242 self.declare_artifact(self.source.url_path, sources=list(self.source.iter_source_filenames()))
244 def build_artifact(self, artifact):
245 with artifact.open("wb") as df:
246 with open(self.source.attachment_filename, "rb") as sf:
247 shutil.copyfileobj(sf, df)
250@buildprogram(File)
251class FileAssetBuildProgram(BuildProgram):
252 def produce_artifacts(self):
253 self.declare_artifact(self.source.artifact_name, sources=[self.source.source_filename])
255 def build_artifact(self, artifact):
256 with artifact.open("wb") as df:
257 with open(self.source.source_filename, "rb") as sf:
258 shutil.copyfileobj(sf, df)
261@buildprogram(Directory)
262class DirectoryAssetBuildProgram(BuildProgram):
263 def iter_child_sources(self):
264 return self.source.children