Coverage for src/lektor_ng/types/flow.py: 80%
143 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-03 22:08 +0000
1import re
3from jinja2 import TemplateNotFound, is_undefined
4from markupsafe import Markup
6from lektor_ng.constants import PRIMARY_ALT
7from lektor_ng.context import get_ctx
8from lektor_ng.metaformat import tokenize
9from lektor_ng.types.base import Type
11_block_re = re.compile(r"^####\s*([^#]*?)\s*####\s*$")
12_line_unescape_re = re.compile(r"^#####(.*?)#####(\s*)$")
15def discover_relevant_flowblock_models(flow, pad, record, alt):
16 """Returns a dictionary of all relevant flow blocks. If no list of
17 flow block names is provided all flow blocks are returned. Otherwise
18 only flow blocks that are in the list or are children of flowblocks
19 in the list are returned.
20 """
21 flow_blocks = flow.flow_blocks
23 all_blocks = pad.db.flowblocks
24 if flow_blocks is None:
25 return {k: v.to_json(pad, record, alt) for k, v in all_blocks.items()}
27 wanted_blocks = set()
28 to_process = flow_blocks[:]
30 while to_process:
31 block_name = to_process.pop()
32 flowblock = all_blocks.get(block_name)
33 if block_name in wanted_blocks or flowblock is None:
34 continue
35 wanted_blocks.add(block_name)
36 for field in flowblock.fields:
37 if isinstance(field.type, FlowType):
38 if field.type.flow_blocks is None:
39 raise RuntimeError("Nested flow-blocks require explicit list of involved blocks.")
40 to_process.extend(field.type.flow_blocks)
42 rv = {}
43 for block_name in wanted_blocks:
44 rv[block_name] = all_blocks[block_name].to_json(pad, record, alt)
46 return rv
49class BadFlowBlock(Exception):
50 pass
53class FlowBlock:
54 """Represents a flowblock for the template."""
56 def __init__(self, data, pad, record):
57 self._data = data
58 self._bound_data = {}
59 self.pad = pad
60 self.record = record
62 @property
63 def flowblockmodel(self):
64 """The flowblock model that created this flow block."""
65 return self.pad.db.flowblocks[self._data["_flowblock"]]
67 def __contains__(self, name):
68 return name in self._data and not is_undefined(self._data[name])
70 def __getitem__(self, name):
71 # If any data of a flowblock is accessed, we record that we need
72 # this dependency.
73 ctx = get_ctx()
74 if ctx is not None:
75 ctx.record_dependency(self.flowblockmodel.filename)
77 rv = self._bound_data.get(name, Ellipsis)
78 if rv is not Ellipsis:
79 return rv
80 rv = self._data[name]
81 if hasattr(rv, "__get__"):
82 rv = rv.__get__(self.record)
83 self._bound_data[name] = rv
84 return rv
86 def __html__(self):
87 ctx = get_ctx()
89 # If we're in a nested render, we disable the rendering here or we
90 # risk a recursion error.
91 if ctx is None or self in ctx.flow_block_render_stack:
92 return Markup.escape(repr(self))
94 ctx.flow_block_render_stack.append(self)
95 try:
96 try:
97 return self.pad.db.env.render_template(
98 [
99 "blocks/{}.html".format(self._data["_flowblock"]),
100 "blocks/default.html",
101 ],
102 pad=self.pad,
103 this=self,
104 alt=self.record.alt,
105 values={"record": self.record},
106 )
107 except TemplateNotFound:
108 return Markup("[could not find snippet template]")
109 finally:
110 ctx.flow_block_render_stack.pop()
112 def __repr__(self):
113 return f"<{self.__class__.__name__} {self['_flowblock']!r}>"
116class Flow:
117 def __init__(self, blocks, record):
118 self.blocks = blocks
119 self.record = record
121 def __html__(self):
122 return Markup("\n\n".join(x.__html__() for x in self.blocks))
124 def __bool__(self):
125 return bool(self.blocks)
127 __nonzero__ = __bool__
129 def __repr__(self):
130 return f"<{self.__class__.__name__} {self.blocks!r}>"
133class FlowDescriptor:
134 def __init__(self, blocks, pad):
135 self._blocks = blocks
136 self._pad = pad
138 def __get__(self, obj, type=None):
139 if obj is None:
140 return self
141 return Flow([FlowBlock(data, self._pad, obj) for data in self._blocks], obj)
144def process_flowblock_data(raw_value):
145 lineiter = iter(raw_value.splitlines(True))
146 block = None
147 buf = []
148 blocks = []
150 for line in lineiter:
151 # Until we found the first block, we ignore leading whitespace.
152 if block is None and not line.strip():
153 continue
155 # Find a new block start
156 block_start = _block_re.match(line)
157 if block_start is None:
158 if block is None:
159 raise BadFlowBlock("Did not find beginning of flow block")
160 else:
161 if block is not None:
162 blocks.append((block, buf))
163 buf = []
164 block = block_start.group(1)
165 continue
166 buf.append(_line_unescape_re.sub("####\\1####\\2", line))
168 if block is not None:
169 blocks.append((block, buf))
171 return blocks
174class FlowType(Type):
175 widget = "flow"
177 def __init__(self, env, options):
178 Type.__init__(self, env, options)
179 self.flow_blocks = [x.strip() for x in options.get("flow_blocks", "").split(",") if x.strip()] or None
181 def value_from_raw(self, raw):
182 if raw.value is None:
183 return raw.missing_value("Missing flow")
184 if raw.pad is None:
185 return raw.missing_value("Flow value was technically present but used in a place where it cannot be used.")
187 db = raw.pad.db
188 rv = []
190 try:
191 for block, block_lines in process_flowblock_data(raw.value):
192 # Unknown flow blocks are skipped for the moment
193 if self.flow_blocks is not None and block not in self.flow_blocks:
194 continue
195 flowblock = db.flowblocks.get(block)
196 if flowblock is None:
197 continue
199 d = {}
200 for key, lines in tokenize(block_lines):
201 d[key] = "".join(lines)
202 rv.append(flowblock.process_raw_data(d, pad=raw.pad))
203 except BadFlowBlock as e:
204 return raw.bad_value(str(e))
206 return FlowDescriptor(rv, raw.pad)
208 def to_json(self, pad, record=None, alt=PRIMARY_ALT):
209 rv = Type.to_json(self, pad, record, alt)
211 rv["flowblocks"] = discover_relevant_flowblock_models(self, pad, record, alt)
213 block_order = self.flow_blocks
214 if block_order is None:
215 block_order = [k for k, v in sorted(pad.db.flowblocks.items(), key=lambda x: x[1].order)]
216 rv["flowblock_order"] = block_order
218 return rv