Coverage for src/lektor_ng/editor.py: 83%

1import os

2import posixpath

3import shutil

4import warnings

5from collections import ChainMap, OrderedDict

6from collections.abc import ItemsView, KeysView, Mapping, MutableMapping, ValuesView

7from contextlib import suppress

8from functools import wraps

9from itertools import chain

11from lektor_ng.constants import PRIMARY_ALT

12from lektor_ng.metaformat import serialize

13from lektor_ng.utils import (

14 atomic_open,

15 cleanup_path,

16 increment_filename,

17 is_valid_id,

18 parse_path,

19 secure_filename,

20)

22implied_keys = {"_id", "_path", "_gid", "_alt", "_source_alt", "_attachment_for"}

23possibly_implied_keys = {"_model", "_template", "_attachment_type"}

26class BadEdit(Exception):

27 pass

30class BadDelete(BadEdit):

31 pass

34def _is_valid_path(path: str) -> bool:

35 split_path = path.strip("/").split("/")

36 if split_path == [""]:

37 split_path = []

38 return parse_path(path) == split_path

41def make_editor_session(pad, path, is_attachment=None, alt=PRIMARY_ALT, datamodel=None):

42 """Creates an editor session for the given path object."""

43 if not _is_valid_path(path):

44 raise BadEdit("Invalid path")

45 path = cleanup_path(path)

47 if alt != PRIMARY_ALT and not pad.db.config.is_valid_alternative(alt):

48 raise BadEdit(f"Attempted to edit an invalid alternative ({alt})")

50 raw_data = pad.db.load_raw_data(path, cls=OrderedDict, alt=alt, fallback=False)

51 raw_data_fallback = None

52 if alt != PRIMARY_ALT:

53 raw_data_fallback = pad.db.load_raw_data(path, cls=OrderedDict)

54 all_data = OrderedDict()

55 all_data.update(raw_data_fallback or ())

56 all_data.update(raw_data or ())

57 else:

58 all_data = raw_data

60 id = posixpath.basename(path)

61 if not is_valid_id(id):

62 raise BadEdit("Invalid ID")

64 record = None

65 exists = raw_data is not None or raw_data_fallback is not None

66 if raw_data is None:

67 raw_data = OrderedDict()

69 if is_attachment is None:

70 if not exists:

71 is_attachment = False

72 else:

73 is_attachment = bool(all_data.get("_attachment_for"))

74 elif bool(all_data.get("_attachment_for")) != is_attachment:

75 raise BadEdit("The attachment flag passed is conflicting with the record's attachment flag.")

77 if exists:

78 # XXX: what about changing the datamodel after the fact?

79 if datamodel is not None:

80 raise BadEdit("When editing an existing record, a datamodel must not be provided.")

81 datamodel = pad.db.get_datamodel_for_raw_data(all_data, pad)

82 else:

83 if datamodel is None:

84 datamodel = pad.db.get_implied_datamodel(path, is_attachment, pad)

85 elif isinstance(datamodel, str):

86 datamodel = pad.db.datamodels[datamodel]

88 if exists:

89 record = pad.instance_from_data(dict(all_data), datamodel)

91 for key in implied_keys:

92 raw_data.pop(key, None)

93 if raw_data_fallback:

94 raw_data_fallback.pop(key, None)

96 return EditorSession(

97 pad,

98 id,

99 str(path),

100 original_data=raw_data,

101 fallback_data=raw_data_fallback,

102 datamodel=datamodel,

103 record=record,

104 exists=exists,

105 is_attachment=is_attachment,

106 alt=alt,

107 )

108

109

110def _deprecated_data_proxy(wrapped):

111 """Issue warning when deprecated mapping methods are used directly on

112 EditorSession.

113 """

114

115 name = wrapped.__name__

116 newname = name.removeprefix("iter")

117

118 @wraps(wrapped)

119 def wrapper(self, *args, **kwargs):

120 warnings.warn(

121 f"EditorSession.{name} has been deprecated as of Lektor 3.3.2. "

122 f"Please use EditorSession.data.{newname} instead.",

123 DeprecationWarning,

124 stacklevel=2,

125 )

126 return wrapped(self, *args, **kwargs)

127

128 return wrapper

129

130

131class EditorSession:

132 def __init__(

133 self,

134 pad,

135 id,

136 path,

137 *,

138 original_data,

139 fallback_data,

140 datamodel,

141 record,

142 exists=True,

143 is_attachment=False,

144 alt=PRIMARY_ALT,

145 ):

146 self.id = id

147 self.pad = pad

148 self.path = path

149 self.record = record

150 self.exists = exists

151 self.datamodel = datamodel

152 self.is_root = path.strip("/") == ""

153 self.alt = alt

154

155 slug_format = None

156 parent_name = posixpath.dirname(path)

157 if parent_name != path:

158 parent = pad.get(parent_name)

159 if parent is not None:

160 slug_format = parent.datamodel.child_config.slug_format

161 if slug_format is None:

162 slug_format = "{{ this._id }}"

163 self.slug_format = slug_format

164 self.implied_attachment_type = None

165

166 if is_attachment:

167 self.implied_attachment_type = pad.db.get_attachment_type(path)

168

169 self.data = MutableEditorData(original_data, fallback_data)

170

171 self._delete_this = False

172 self._recursive_delete = False

173 self._master_delete = False

174 self.is_attachment = is_attachment

175 self.closed = False

176

177 def to_json(self):

178 label = None

179 label_i18n = None

180 url_path = None

181 if self.record is not None:

182 label = self.record.record_label

183 label_i18n = self.record.get_record_label_i18n()

184 url_path = self.record.url_path

185 else:

186 label = self.id

187 can_be_deleted = not self.datamodel.protected and not self.is_root

188 return {

189 "data": dict(self.data.items()),

190 "record_info": {

191 "id": self.id,

192 "path": self.path,

193 "exists": self.exists,

194 "label": label,

195 "label_i18n": label_i18n,

196 "url_path": url_path,

197 "alt": self.alt,

198 "is_attachment": self.is_attachment,

199 "can_be_deleted": can_be_deleted,

200 "slug_format": self.slug_format,

201 "implied_attachment_type": self.implied_attachment_type,

202 "default_template": self.datamodel.get_default_template_name(),

203 },

204 "datamodel": self.datamodel.to_json(self.pad, self.record, self.alt),

205 }

206

207 def __enter__(self):

208 return self

209

210 def __exit__(self, exc_type, exc_value, tb):

211 if exc_type is not None:

212 self.rollback()

213 else:

214 self.commit()

215

216 def get_fs_path(self, alt=PRIMARY_ALT):

217 """The path to the record file on the file system."""

218 base = self.pad.db.to_fs_path(self.path)

219 suffix = ".lr"

220 if alt != PRIMARY_ALT:

221 suffix = f"+{alt}{suffix}"

222 if self.is_attachment:

223 return base + suffix

224 return os.path.join(base, "contents" + suffix)

225

226 @property

227 def fs_path(self):

228 """The file system path of the content file on disk."""

229 return self.get_fs_path(self.alt)

230

231 @property

232 def attachment_fs_path(self):

233 """The file system path of the actual attachment."""

234 if self.is_attachment:

235 return self.pad.db.to_fs_path(self.path)

236 return None

237

238 def rollback(self):

239 """Ignores all changes and rejects them."""

240 if self.closed:

241 return

242 self.closed = True

243

244 def commit(self):

245 """Saves changes back to the file system."""

246 if not self.closed:

247 if self._delete_this:

248 self._delete_impl()

249 else:

250 self._save_impl()

251 self.closed = True

252

253 def delete(self, recursive=None, delete_master=False):

254 """Deletes the record. How the delete works depends on what is being

255 deleted:

256

257 * delete attachment: recursive mode is silently ignored. If

258 `delete_master` is set then the attachment is deleted, otherwise

259 only the metadata is deleted.

260 * delete page: in recursive mode everything is deleted in which

261 case `delete_master` must be set to `True` or an error is

262 generated. In fact, the default is to perform a recursive

263 delete in that case. If `delete_master` is False, then only the

264 contents file of the current alt is deleted.

265

266 If a delete cannot be performed, an error is generated.

267 """

268 if self.closed:

269 return

270 if recursive is None:

271 recursive = not self.is_attachment and delete_master

272 self._delete_this = True

273 self._recursive_delete = recursive

274 self._master_delete = delete_master

275

276 def add_attachment(self, filename, fp):

277 """Stores a new attachment. Returns `None` if the file already"""

278 if not self.exists:

279 raise BadEdit("Record does not exist.")

280 if self.is_attachment:

281 raise BadEdit("Cannot attach something to an attachment.")

282 if not self.datamodel.has_own_attachments:

283 raise BadEdit("Attachments are disabled for this page.")

284 directory = self.pad.db.to_fs_path(self.path)

285

286 safe_filename = secure_filename(filename)

287

288 while 1:

289 fn = os.path.join(directory, safe_filename)

290 if not os.path.isfile(fn):

291 break

292 safe_filename = increment_filename(fn)

293

294 with atomic_open(fn, "wb") as f:

295 shutil.copyfileobj(fp, f)

296 return safe_filename

297

298 def _attachment_delete_impl(self):

299 files = [self.fs_path]

300 if self._master_delete:

301 files.append(self.attachment_fs_path)

302 files.extend(self.get_fs_path(alt) for alt in self.pad.db.config.list_alternatives())

303 for fn in files:

304 try:

305 os.unlink(fn)

306 except OSError:

307 pass

308

309 def _page_delete_impl(self):

310 directory = os.path.dirname(self.fs_path)

311

312 if self._recursive_delete:

313 with suppress(OSError):

314 shutil.rmtree(directory)

315 return

316 if self._master_delete:

317 raise BadDelete("Master deletes of pages require that recursive deleting is enabled.")

318

319 for fn in self.fs_path, directory:

320 try:

321 os.unlink(fn)

322 except OSError:

323 pass

324

325 def _delete_impl(self):

326 if self.alt != PRIMARY_ALT:

327 if self._master_delete:

328 raise BadDelete(

329 f'Master deletes need to be done from the primary alt. Tried to delete from "{self.alt}"'

330 )

331 if self._recursive_delete:

332 raise BadDelete(

333 f'Cannot perform recursive delete from a non primary alt. Tried to delete from "{self.alt}"'

334 )

335

336 if self.is_attachment:

337 self._attachment_delete_impl()

338 else:

339 self._page_delete_impl()

340

341 def _save_impl(self):

342 # When creating a new alt from a primary self.exists is True but

343 # the file does not exist yet. In this case we want to explicitly

344 # create it anyways instead of bailing.

345 if not self.data.ischanged() and self.exists and os.path.exists(self.fs_path):

346 return

347

348 try:

349 os.makedirs(os.path.dirname(self.fs_path))

350 except OSError:

351 pass

352

353 with atomic_open(self.fs_path, "wb") as f:

354 for chunk in serialize(self.data.items(fallback=False), encoding="utf-8"):

355 f.write(chunk)

356

357 def __repr__(self):

358 bits = [repr(self.path)]

359 if self.alt != PRIMARY_ALT:

360 bits.append(f" alt={self.alt!r}")

361 if not self.exists:

362 bits.append(" new")

363

364 return f"<{self.__class__.__name__} {' '.join(bits)}>"

365

366 # The mapping methods used to access the page data have been moved

367 # to EditorSession.data.

368 #

369 # We have left behind these proxy methods so as not to break any existing

370 # external code.

371 @_deprecated_data_proxy

372 def revert_key(self, key):

373 self.data.revert_key(key)

374

375 @_deprecated_data_proxy

376 def __contains__(self, key):

377 return key in self.data

378

379 @_deprecated_data_proxy

380 def __getitem__(self, key):

381 return self.data[key]

382

383 @_deprecated_data_proxy

384 def __len__(self):

385 return len(self.data)

386

387 @_deprecated_data_proxy

388 def __iter__(self):

389 return iter(self.data)

390

391 @_deprecated_data_proxy

392 def items(self, fallback=True):

393 return self.data.items(fallback)

394

395 @_deprecated_data_proxy

396 def keys(self, fallback=True):

397 return self.data.keys(fallback)

398

399 @_deprecated_data_proxy

400 def values(self, fallback=True):

401 return self.data.values(fallback)

402

403 @_deprecated_data_proxy

404 def iteritems(self, fallback=True):

405 return self.data.items(fallback)

406

407 @_deprecated_data_proxy

408 def iterkeys(self, fallback=True):

409 return self.data.keys(fallback)

410

411 @_deprecated_data_proxy

412 def itervalues(self, fallback=True):

413 return self.data.values(fallback)

414

415 @_deprecated_data_proxy

416 def __setitem__(self, key, value):

417 self.data[key] = value

418

419 @_deprecated_data_proxy

420 def update(self, *args, **kwargs):

421 self.data.update(*args, **kwargs)

422

423

424del _deprecated_data_proxy

425

426

427class EditorData(Mapping):

428 """A read-only view of edited data.

429

430 This is a chained dict with (possibly) mutated data overlaid on

431 the original data for the record.

432 """

433

434 def __init__(self, original_data, fallback_data=None, _changed_data=None):

435 if _changed_data is None:

436 _changed_data = {}

437 self.fallback_data = fallback_data

438 if fallback_data:

439 self._orig_data = ChainMap(original_data, fallback_data)

440 self._data = ChainMap(_changed_data, original_data, fallback_data)

441 else:

442 self._orig_data = original_data

443 self._data = ChainMap(_changed_data, original_data)

444

445 @property

446 def _changed_data(self):

447 return self._data.maps[0]

448

449 @property

450 def original_data(self):

451 return self._data.maps[1]

452

453 def ischanged(self):

454 return len(self._changed_data) > 0

455

456 def __getitem__(self, key):

457 rv = self._data.get(key)

458 if rv is None:

459 raise KeyError(key)

460 return rv

461

462 def __iter__(self):

463 data = self._data

464 fallback_data = self.fallback_data or {}

465

466 for key in _uniq(chain(self.original_data, fallback_data, sorted(data))):

467 if key not in implied_keys:

468 if data[key] is not None:

469 yield key

470

471 def __len__(self):

472 data = self._data

473 return sum(1 for key in data if key not in implied_keys and data[key] is not None)

474

475 def keys(self, fallback=True): # pylint: disable=arguments-differ

476 return KeysView(self if fallback else self._without_fallback())

477

478 def items(self, fallback=True): # pylint: disable=arguments-differ

479 return ItemsView(self if fallback else self._without_fallback())

480

481 def values(self, fallback=True): # pylint: disable=arguments-differ

482 return ValuesView(self if fallback else self._without_fallback())

483

484 def _without_fallback(self):

485 """Return a copy of self, with fallback_data set to None."""

486 if not self.fallback_data:

487 return self

488 return EditorData(self.original_data, _changed_data=self._changed_data)

489

490

491class MutableEditorData(EditorData, MutableMapping):

492 """A mutable view of edited data.

493

494 This is a chained dict with (possibly) mutated data overlaid on

495 the original data for the record.

496 """

497

498 def __setitem__(self, key, value):

499 if key in implied_keys:

500 raise KeyError(f"Can not set implied key {key!r}")

501 orig_value = self._orig_data.get(key)

502 if value != orig_value:

503 self._data[key] = value

504 elif key in possibly_implied_keys:

505 # If the key is in the possibly implied key set and set to

506 # that value, we will set it to changed anyways. This allows

507 # overriding of such special keys.

508 self._data[key] = value

509 else:

510 self._data.pop(key, None)

511

512 def __delitem__(self, key):

513 if key in implied_keys or self._data.get(key) is None:

514 raise KeyError(key)

515 self[key] = None

516

517 def revert_key(self, key):

518 """Reverts a key to the implied value."""

519 self._data.pop(key, None)

520

521

522def _uniq(seq):

523 """Return items from iterable in order, skipping items that have already been seen.

524

525 The items in ``seq`` must be hashable.

526 """

527 seen = set()

528 for item in seq:

529 if item not in seen:

530 seen.add(item)

531 yield item