Coverage for src/lektor_ng/datamodel.py: 94%

1import errno

2import os

4from lektor_ng.constants import PRIMARY_ALT

5from lektor_ng.environment.expressions import Expression, FormatExpression

6from lektor_ng.i18n import generate_i18n_kvs, get_i18n_block

7from lektor_ng.inifile import IniFile

8from lektor_ng.pagination import Pagination

9from lektor_ng.reporter import reporter

10from lektor_ng.types import builtin_types

11from lektor_ng.types.base import RawValue

12from lektor_ng.utils import bool_from_string, slugify

15class ChildConfig:

16 def __init__(

17 self,

18 *,

19 enabled=None,

20 slug_format=None,

21 model=None,

22 order_by=None,

23 replaced_with=None,

24 hidden=None,

25 ):

26 if enabled is None:

27 enabled = True

28 self.enabled = enabled

29 self.slug_format = slug_format

30 self.model = model

31 self.order_by = order_by

32 self.replaced_with = replaced_with

33 self.hidden = hidden

35 def to_json(self):

36 return {

37 "enabled": self.enabled,

38 "slug_format": self.slug_format,

39 "model": self.model,

40 "order_by": self.order_by,

41 "replaced_with": self.replaced_with,

42 "hidden": self.hidden,

43 }

46class PaginationConfig:

47 def __init__(self, env, *, enabled=None, per_page=None, url_suffix=None, items=None):

48 self.env = env

49 if enabled is None:

50 enabled = False

51 self.enabled = enabled

52 if per_page is None:

53 per_page = 20

54 elif not isinstance(per_page, int):

55 raise TypeError(f"per_page must be an int or None, not {per_page!r}")

56 if per_page <= 0:

57 raise ValueError("per_page must be positive, not {per_page}")

59 self.per_page = per_page

60 if url_suffix is None:

61 url_suffix = "page"

62 self.url_suffix = url_suffix

63 self.items = items

64 self._items_tmpl = None

66 def count_total_items(self, record):

67 """Counts the number of items over all pages."""

68 return self.get_pagination_query(record).count()

70 def count_pages(self, record):

71 """Returns the total number of pages for the children of a record."""

72 total = self.count_total_items(record)

73 npages = (total + self.per_page - 1) // self.per_page

74 # Even when there are no children, we want at least one page

75 return max(npages, 1)

77 def slice_query_for_page(self, record, page):

78 """Slices the query so it returns the children for a given page."""

79 query = self.get_pagination_query(record)

80 if not self.enabled or page is None:

81 return query

82 return query.limit(self.per_page).offset((page - 1) * self.per_page)

84 @staticmethod

85 def get_record_for_page(record, page_num):

86 """Given a normal record this one returns the version specific

87 for a page.

88 """

89 # If we already have the right version, return it.

90 if record.page_num == page_num:

91 return record

93 # Check if we have a cached version

94 pad = record.pad

95 rv = pad.cache.get(record.path, record.alt, str(page_num))

96 if rv is not Ellipsis:

97 return rv

99 # Make what we need out of what we have and put it into the cache.

100 cls = record.__class__

101 rv = cls(record.pad, record._data, page_num=page_num)

102 pad.cache.remember(rv)

103 return rv

104

105 def match_pagination(self, record, url_path):

106 """Matches the pagination from the URL path."""

107 if not self.enabled:

108 return None

109 suffixes = self.url_suffix.strip("/").split("/")

110 if url_path[: len(suffixes)] != suffixes:

111 return None

112 try:

113 page_num = int(url_path[len(suffixes)])

114 except (ValueError, IndexError):

115 return None

116

117 # It's important we do not allow "1" here as the first page is always

118 # on the root. Changing this would mean the URLs are incorrectly

119 # generated if someone manually went to /page/1/.

120 if page_num == 1 or len(url_path) != len(suffixes) + 1:

121 return None

122

123 # Page needs to have at least a single child.

124 rv = self.get_record_for_page(record, page_num)

125 if rv.pagination.items.first() is not None:

126 return rv

127 return None

128

129 def get_pagination_controller(self, record):

130 if not self.enabled:

131 raise RuntimeError("Pagination is disabled")

132 return Pagination(record, self)

133

134 def get_pagination_query(self, record):

135 items_expr = self.items

136 if items_expr is None:

137 return record.children

138 if self._items_tmpl is None or self._items_tmpl[0] != items_expr:

139 self._items_tmpl = (items_expr, Expression(self.env, items_expr))

140

141 return self._items_tmpl[1].evaluate(record.pad, this=record)

142

143 def to_json(self):

144 return {

145 "enabled": self.enabled,

146 "per_page": self.per_page,

147 "url_suffix": self.url_suffix,

148 "items": self.items,

149 }

150

151

152class AttachmentConfig:

153 def __init__(self, *, enabled=None, model=None, order_by=None, hidden=None):

154 if enabled is None:

155 enabled = True

156 if hidden is None:

157 hidden = False

158 self.enabled = enabled

159 self.model = model

160 self.order_by = order_by

161 self.hidden = hidden

162

163 def to_json(self):

164 return {

165 "enabled": self.enabled,

166 "model": self.model,

167 "order_by": self.order_by,

168 "hidden": self.hidden,

169 }

170

171

172class Field:

173 def __init__(self, env, name, type=None, options=None):

174 if type is None:

175 type = env.types["string"]

176 if options is None:

177 options = {}

178 self.options = options

179 self.name = name

180 label_i18n = get_i18n_block(options, "label")

181 if not label_i18n:

182 label_i18n = {"en": name.replace("_", " ").strip().capitalize()}

183 self.label_i18n = label_i18n

184 self.description_i18n = get_i18n_block(options, "description") or None

185 self.default = options.get("default")

186 self.type = type(env, options)

187

188 @property

189 def label(self):

190 return self.label_i18n.get("en")

191

192 def to_json(self, pad, record=None, alt=PRIMARY_ALT):

193 return {

194 "name": self.name,

195 "label": self.label,

196 "label_i18n": self.label_i18n,

197 "hide_label": bool_from_string(self.options.get("hide_label"), default=False),

198 "description_i18n": self.description_i18n,

199 "type": self.type.to_json(pad, record, alt),

200 "default": self.default,

201 "alts_enabled": bool_from_string(self.options.get("alts_enabled"), default=None),

202 }

203

204 def deserialize_value(self, value, pad=None):

205 raw_value = RawValue(self.name, value, field=self, pad=pad)

206 return self.type.value_from_raw_with_default(raw_value)

207

208 def serialize_value(self, value):

209 return self.type.value_to_raw(value)

210

211 def __repr__(self):

212 return f"<{self.__class__.__name__} {self.name!r} type={self.type!r}>"

213

214

215def _iter_all_fields(obj):

216 for name in sorted(x for x in obj.field_map if x[:1] == "_"):

217 yield obj.field_map[name]

218 yield from obj.fields

219

220

221class DataModel:

222 def __init__(

223 self,

224 env,

225 id,

226 name_i18n,

227 *,

228 label_i18n=None,

229 filename=None,

230 hidden=None,

231 protected=None,

232 child_config=None,

233 attachment_config=None,

234 pagination_config=None,

235 fields=None,

236 primary_field=None,

237 parent=None,

238 ):

239 self.env = env

240 self.filename = filename

241 self.id = id

242 self.name_i18n = name_i18n

243 self.label_i18n = label_i18n

244 if hidden is None:

245 hidden = False

246 self.hidden = hidden

247 if protected is None:

248 protected = False

249 self.protected = protected

250 if child_config is None:

251 child_config = ChildConfig()

252 self.child_config = child_config

253 if attachment_config is None:

254 attachment_config = AttachmentConfig()

255 self.attachment_config = attachment_config

256 if pagination_config is None:

257 pagination_config = PaginationConfig(env)

258 self.pagination_config = pagination_config

259 if fields is None:

260 fields = []

261 self.fields = fields

262 if primary_field is None and fields:

263 primary_field = fields[0].name

264 self.primary_field = primary_field

265 self.parent = parent

266

267 # This is a mapping of the key names to the actual field which

268 # also includes the system fields. This is primarily used for

269 # fast internal operations but also the admin.

270 self.field_map = {x.name: x for x in fields}

271 for key, (ty, opts) in system_fields.items():

272 self.field_map[key] = Field(env, name=key, type=ty, options=opts)

273

274 self._child_slug_tmpl = None

275 self._child_replacements = None

276 self._label_tmpls = {}

277

278 @property

279 def name(self):

280 name = (self.name_i18n or {}).get("en")

281 return name or self.id.title().replace("_", " ")

282

283 @property

284 def label(self):

285 return (self.label_i18n or {}).get("en")

286

287 def to_json(self, pad, record=None, alt=PRIMARY_ALT):

288 """Describes the datamodel as JSON data."""

289 return {

290 "filename": self.filename,

291 "alt": alt,

292 "id": self.id,

293 "name": self.name,

294 "name_i18n": self.name_i18n,

295 "primary_field": self.primary_field,

296 "label": self.label,

297 "label_i18n": self.label_i18n,

298 "hidden": self.hidden,

299 "protected": self.protected,

300 "child_config": self.child_config.to_json(),

301 "attachment_config": self.attachment_config.to_json(),

302 "pagination_config": self.pagination_config.to_json(),

303 "fields": [x.to_json(pad, record, alt) for x in _iter_all_fields(self)],

304 }

305

306 def format_record_label(self, record, lang="en"):

307 """Returns the label for a given record."""

308 label = self.label_i18n.get(lang)

309 if label is None:

310 return None

311

312 tmpl = self._label_tmpls.get(lang)

313 if tmpl is None:

314 tmpl = (label, FormatExpression(self.env, label))

315 self._label_tmpls[lang] = tmpl

316

317 try:

318 return tmpl[1].evaluate(record.pad, this=record)

319 except Exception:

320 # XXX: log

321 return None

322

323 def get_default_child_slug(self, pad, data):

324 """Formats out the child slug."""

325 slug_format = self.child_config.slug_format

326 if slug_format is None:

327 return data["_id"]

328

329 if self._child_slug_tmpl is None or self._child_slug_tmpl[0] != slug_format:

330 self._child_slug_tmpl = (

331 slug_format,

332 FormatExpression(self.env, slug_format),

333 )

334

335 try:

336 return "_".join(self._child_slug_tmpl[1].evaluate(pad, this=data).strip().split()).strip("/")

337 except Exception as exc:

338 reporter.report_generic(f"Failed to expand child slug_format: {exc}")

339 return "temp-" + slugify(data["_id"])

340

341 def get_default_template_name(self):

342 return self.id + ".html"

343

344 @property

345 def has_own_children(self):

346 return self.child_config.replaced_with is None and self.child_config.enabled

347

348 @property

349 def has_own_attachments(self):

350 return self.attachment_config.enabled

351

352 def get_child_replacements(self, record):

353 """Returns the query that should be used as replacement for the

354 actual children.

355 """

356 replaced_with = self.child_config.replaced_with

357 if replaced_with is None:

358 return None

359

360 if self._child_replacements is None or self._child_replacements[0] != replaced_with:

361 self._child_replacements = (

362 replaced_with,

363 Expression(self.env, replaced_with),

364 )

365

366 return self._child_replacements[1].evaluate(record.pad, this=record)

367

368 def process_raw_data(self, raw_data, pad=None):

369 rv = {}

370 for field in self.field_map.values():

371 value = raw_data.get(field.name)

372 rv[field.name] = field.deserialize_value(value, pad=pad)

373 rv["_model"] = self.id

374 return rv

375

376 def __repr__(self):

377 return f"<{self.__class__.__name__} {self.id!r}>"

378

379

380class FlowBlockModel:

381 def __init__(

382 self,

383 env,

384 id,

385 name_i18n,

386 *,

387 filename=None,

388 fields=None,

389 order=None,

390 button_label=None,

391 ):

392 self.env = env

393 self.id = id

394 self.name_i18n = name_i18n

395 self.filename = filename

396 if fields is None:

397 fields = []

398 self.fields = fields

399 if order is None:

400 order = 100

401 self.order = order

402 self.button_label = button_label

403

404 self.field_map = {x.name: x for x in fields}

405 self.field_map["_flowblock"] = Field(env, name="_flowblock", type=env.types["string"])

406

407 @property

408 def name(self):

409 return self.name_i18n.get("en") or self.id.title().replace("_", " ")

410

411 def to_json(self, pad, record=None, alt=PRIMARY_ALT):

412 return {

413 "id": self.id,

414 "name": self.name,

415 "name_i18n": self.name_i18n,

416 "filename": self.filename,

417 "fields": [x.to_json(pad, record, alt) for x in _iter_all_fields(self) if x.name != "_flowblock"],

418 "order": self.order,

419 "button_label": self.button_label,

420 }

421

422 def process_raw_data(self, raw_data, pad=None):

423 rv = {}

424 for field in self.field_map.values():

425 value = raw_data.get(field.name)

426 rv[field.name] = field.deserialize_value(value, pad=pad)

427 rv["_flowblock"] = self.id

428 return rv

429

430 def __repr__(self):

431 return f"<{self.__class__.__name__} {self.id!r}>"

432

433

434def fielddata_from_ini(inifile):

435 return [

436 (

437 sect.split(".", 1)[1],

438 inifile.section_as_dict(sect),

439 )

440 for sect in inifile.sections()

441 if sect.startswith("fields.")

442 ]

443

444

445def datamodel_data_from_ini(id, inifile):

446 def _parse_order(value):

447 value = (value or "").strip()

448 if not value:

449 return None

450 return [x for x in [x.strip() for x in value.strip().split(",")] if x]

451

452 return {

453 "filename": inifile.filename,

454 "id": id,

455 "parent": inifile.get("model.inherits"),

456 "name_i18n": get_i18n_block(inifile, "model.name"),

457 "label_i18n": get_i18n_block(inifile, "model.label"),

458 "primary_field": inifile.get("model.primary_field"),

459 "hidden": inifile.get_bool("model.hidden", default=None),

460 "protected": inifile.get_bool("model.protected", default=None),

461 "child_config": {

462 "enabled": inifile.get_bool("children.enabled", default=None),

463 "slug_format": inifile.get("children.slug_format"),

464 "model": inifile.get("children.model"),

465 "order_by": _parse_order(inifile.get("children.order_by")),

466 "replaced_with": inifile.get("children.replaced_with"),

467 "hidden": inifile.get_bool("children.hidden", default=None),

468 },

469 "attachment_config": {

470 "enabled": inifile.get_bool("attachments.enabled", default=None),

471 "model": inifile.get("attachments.model"),

472 "order_by": _parse_order(inifile.get("attachments.order_by")),

473 "hidden": inifile.get_bool("attachments.hidden", default=None),

474 },

475 "pagination_config": {

476 "enabled": inifile.get_bool("pagination.enabled", default=None),

477 "per_page": inifile.get_int("pagination.per_page"),

478 "url_suffix": inifile.get("pagination.url_suffix"),

479 "items": inifile.get("pagination.items"),

480 },

481 "fields": fielddata_from_ini(inifile),

482 }

483

484

485def flowblock_data_from_ini(id, inifile):

486 return {

487 "filename": inifile.filename,

488 "id": id,

489 "name_i18n": get_i18n_block(inifile, "block.name"),

490 "fields": fielddata_from_ini(inifile),

491 "order": inifile.get_int("block.order"),

492 "button_label": inifile.get("block.button_label"),

493 }

494

495

496def fields_from_data(env, data, parent_fields=None):

497 fields = []

498 known_fields = set()

499

500 for name, options in data:

501 ty = env.types[options.get("type", "string")]

502 fields.append(Field(env=env, name=name, type=ty, options=options))

503 known_fields.add(name)

504

505 if parent_fields is not None:

506 prepended_fields = [field for field in parent_fields if field.name not in known_fields]

507 fields = prepended_fields + fields

508

509 return fields

510

511

512def datamodel_from_data(env, model_data, parent=None):

513 def get_value(key):

514 path = key.split(".")

515 node = model_data

516 for item in path:

517 node = node.get(item)

518 if node is not None:

519 return node

520 if parent is not None:

521 node = parent

522 for item in path:

523 node = getattr(node, item)

524 return node

525 return None

526

527 fields = fields_from_data(env, model_data["fields"], parent and parent.fields or None)

528

529 return DataModel(

530 env,

531 # data that never inherits

532 filename=model_data["filename"],

533 id=model_data["id"],

534 parent=parent,

535 name_i18n=model_data["name_i18n"],

536 primary_field=model_data["primary_field"],

537 # direct data that can inherit

538 label_i18n=get_value("label_i18n"),

539 hidden=get_value("hidden"),

540 protected=get_value("protected"),

541 child_config=ChildConfig(

542 enabled=get_value("child_config.enabled"),

543 slug_format=get_value("child_config.slug_format"),

544 model=get_value("child_config.model"),

545 order_by=get_value("child_config.order_by"),

546 replaced_with=get_value("child_config.replaced_with"),

547 hidden=get_value("child_config.hidden"),

548 ),

549 attachment_config=AttachmentConfig(

550 enabled=get_value("attachment_config.enabled"),

551 model=get_value("attachment_config.model"),

552 order_by=get_value("attachment_config.order_by"),

553 hidden=get_value("attachment_config.hidden"),

554 ),

555 pagination_config=PaginationConfig(

556 env,

557 enabled=get_value("pagination_config.enabled"),

558 per_page=get_value("pagination_config.per_page"),

559 url_suffix=get_value("pagination_config.url_suffix"),

560 items=get_value("pagination_config.items"),

561 ),

562 fields=fields,

563 )

564

565

566def flowblock_from_data(env, block_data):

567 return FlowBlockModel(

568 env,

569 filename=block_data["filename"],

570 id=block_data["id"],

571 name_i18n=block_data["name_i18n"],

572 fields=fields_from_data(env, block_data["fields"]),

573 order=block_data["order"],

574 button_label=block_data["button_label"],

575 )

576

577

578def iter_inis(path):

579 try:

580 for filename in os.listdir(path):

581 if not filename.endswith(".ini") or filename[:1] in "_.":

582 continue

583 fn = os.path.join(path, filename)

584 if os.path.isfile(fn):

585 base = filename[:-4]

586 base = base.encode("utf-8").decode("ascii", "replace")

587 inifile = IniFile(fn)

588 yield base, inifile

589 except OSError as e:

590 if e.errno != errno.ENOENT:

591 raise

592

593

594def load_datamodels(env):

595 """Loads the datamodels for a specific environment."""

596 # Models will override previous loaded models with the same name

597 # So models paths are loaded in reverse order

598 paths = list(reversed(env.theme_paths)) + [env.root_path]

599 paths = [os.path.join(p, "models") for p in paths]

600 data = {}

601

602 for path in paths:

603 for model_id, inifile in iter_inis(path):

604 data[model_id] = datamodel_data_from_ini(model_id, inifile)

605

606 rv = {}

607

608 def get_model(model_id):

609 model = rv.get(model_id)

610 if model is not None:

611 return model

612 if model_id in data:

613 return create_model(model_id)

614 return None

615

616 def create_model(model_id):

617 model_data = data.get(model_id)

618 if model_data is None:

619 raise RuntimeError(f"Model {model_id!r} not found")

620

621 if model_data["parent"] is not None:

622 parent = get_model(model_data["parent"])

623 else:

624 parent = None

625

626 rv[model_id] = mod = datamodel_from_data(env, model_data, parent)

627 return mod

628

629 for model_id in data:

630 get_model(model_id)

631

632 rv["none"] = DataModel(env, "none", {"en": "None"}, hidden=True)

633

634 return rv

635

636

637def load_flowblocks(env):

638 """Loads all the flow blocks for a specific environment."""

639 # Flowblocks will override previous loaded flowblocks with the same name

640 # So paths are loaded in reverse order

641 paths = list(reversed(env.theme_paths)) + [env.root_path]

642 paths = [os.path.join(p, "flowblocks") for p in paths]

643 rv = {}

644

645 for path in paths:

646 for flowblock_id, inifile in iter_inis(path):

647 rv[flowblock_id] = flowblock_from_data(env, flowblock_data_from_ini(flowblock_id, inifile))

648

649 return rv

650

651

652system_fields = {}

653

654

655def add_system_field(name, **opts):

656 opts = dict(generate_i18n_kvs(**opts))

657 ty = builtin_types[opts.pop("type")]

658 system_fields[name] = (ty, opts)

659

660

661# The full path of the record

662add_system_field("_path", type="string")

663

664# The local ID (within a folder) of the record

665add_system_field("_id", type="string")

666

667# The global ID (within a folder) of the record

668add_system_field("_gid", type="string")

669

670# The alt key that identifies this record

671add_system_field("_alt", type="string")

672

673# The alt key for the file that was actually referenced.

674add_system_field("_source_alt", type="string")

675

676# the model that defines the data of the record

677add_system_field("_model", type="string")

678

679# the template that should be used for rendering if not hidden

680add_system_field(

681 "_template",

682 type="string",

683 label_i18n="TEMPLATE",

684 width="1/2",

685 addon_label="[[code]]",

686)

687

688# the slug that should be used for this record. This is added below the

689# slug of the parent.

690add_system_field("_slug", type="slug", label_i18n="URL_SLUG", width="1/2")

691

692# This can be used to hide an individual record.

693add_system_field(

694 "_hidden",

695 type="boolean",

696 label_i18n="HIDE_PAGE",

697 checkbox_label_i18n="HIDE_PAGE_EXPLANATION",

698)

699

700# This marks a page as undiscoverable.

701add_system_field(

702 "_discoverable",

703 type="boolean",

704 default="yes",

705 label_i18n="PAGE_IS_DISCOVERABLE",

706 checkbox_label_i18n="PAGE_IS_DISCOVERABLE_EXPLANATION",

707)

708

709# Useful fields for attachments.

710add_system_field("_attachment_for", type="string")

711add_system_field(

712 "_attachment_type",

713 type="string",

714 label_i18n="ATTACHMENT_TYPE",

715 addon_label="[[paperclip]]",

716)