Coverage for src/lektor_ng/utils.py: 81%

545 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-08-08 00:39 +0000

1from __future__ import annotations 

2 

3import codecs 

4import io 

5import json 

6import os 

7import posixpath 

8import re 

9import subprocess 

10import sys 

11import tempfile 

12import threading 

13import unicodedata 

14import urllib.parse 

15import uuid 

16import warnings 

17from collections.abc import Callable, Hashable, Iterable, Iterator 

18from contextlib import contextmanager, suppress 

19from dataclasses import dataclass 

20from datetime import datetime 

21from functools import cache, wraps 

22from pathlib import Path, PurePosixPath 

23from typing import IO, TYPE_CHECKING, Any, ClassVar, Literal, TypeVar, overload 

24 

25from jinja2 import is_undefined 

26from markupsafe import Markup 

27from slugify import slugify as _slugify 

28from werkzeug.http import http_date 

29from werkzeug.urls import iri_to_uri, uri_to_iri 

30 

31if TYPE_CHECKING: 

32 from _typeshed import StrPath 

33 

34 

35is_windows = os.name == "nt" 

36 

37_slash_escape = "\\/" not in json.dumps("/") 

38 

39_last_num_re = re.compile(r"^(.*)(\d+)(.*?)$") 

40_list_marker = object() 

41_value_marker = object() 

42 

43# Figure out our fs encoding, if it's ascii we upgrade to utf-8 

44fs_enc = sys.getfilesystemencoding() 

45try: 

46 if codecs.lookup(fs_enc).name == "ascii": 

47 fs_enc = "utf-8" 

48except LookupError: 

49 pass 

50 

51 

52def split_virtual_path(path): 

53 if "@" in path: 

54 return path.split("@", 1) 

55 return path, None 

56 

57 

58def _norm_join(a, b): 

59 return posixpath.normpath(posixpath.join(a, b)) 

60 

61 

62def join_path(a, b): 

63 """Join two DB-paths. 

64 

65 It is assumed that both paths are already normalized in that 

66 neither contains an extra "." or ".." components, double-slashes, 

67 etc. 

68 """ 

69 # NB: This function is really only during URL resolution. The only 

70 # place that references it is lektor.source.SourceObject._resolve_url. 

71 

72 if posixpath.isabs(b): 

73 return b 

74 

75 a_p, a_v = split_virtual_path(a) 

76 b_p, b_v = split_virtual_path(b) 

77 

78 # Special case: paginations are considered special virtual paths 

79 # where the parent is the actual parent of the page. This however 

80 # is explicitly not done if the path we join with refers to the 

81 # current path (empty string or dot). 

82 if b_p not in ("", ".") and a_v and a_v.isdigit(): 

83 a_v = None 

84 

85 # New path has a virtual path, add that to it. 

86 if b_v: 

87 rv = _norm_join(a_p, b_p) + "@" + b_v 

88 elif a_v: 

89 rv = a_p + "@" + _norm_join(a_v, b_p) 

90 else: 

91 rv = _norm_join(a_p, b_p) 

92 if rv[-2:] == "@.": 

93 rv = rv[:-2] 

94 return rv 

95 

96 

97def cleanup_path(path): 

98 # NB: POSIX allows for two leading slashes in a pathname, so we have to 

99 # deal with the possiblity of leading double-slash ourself. 

100 return posixpath.normpath("/" + path.lstrip("/")) 

101 

102 

103def cleanup_url_path(url_path): 

104 """Clean up a URL path. 

105 

106 This strips any query, and/or fragment that may be present in the 

107 input path. 

108 

109 Raises ValueError if the path contains a _scheme_ 

110 which is neither ``http`` nor ``https``, or a _netloc_. 

111 """ 

112 scheme, netloc, path, _, _ = urllib.parse.urlsplit(url_path, scheme="http") 

113 if scheme not in ("http", "https"): 

114 raise ValueError(f"Invalid scheme: {url_path!r}") 

115 if netloc: 

116 raise ValueError(f"Invalid netloc: {url_path!r}") 

117 

118 # NB: POSIX allows for two leading slashes in a pathname, so we have to 

119 # deal with the possiblity of leading double-slash ourself. 

120 return posixpath.normpath("/" + path.lstrip("/")) 

121 

122 

123def parse_path(path): 

124 x = cleanup_path(path).strip("/").split("/") 

125 if x == [""]: 

126 return [] 

127 return x 

128 

129 

130def is_path_child_of(a, b, strict=True): 

131 a_p, a_v = split_virtual_path(a) 

132 b_p, b_v = split_virtual_path(b) 

133 a_p = parse_path(a_p) 

134 b_p = parse_path(b_p) 

135 a_v = parse_path(a_v or "") 

136 b_v = parse_path(b_v or "") 

137 

138 if not strict and a_p == b_p and a_v == b_v: 

139 return True 

140 if not a_v and b_v: 

141 return False 

142 if a_p == b_p and a_v[: len(b_v)] == b_v and len(a_v) > len(b_v): 

143 return True 

144 return a_p[: len(b_p)] == b_p and len(a_p) > len(b_p) 

145 

146 

147def untrusted_to_os_path(path): 

148 if not isinstance(path, str): 

149 path = path.decode(fs_enc, "replace") 

150 clean_path = cleanup_path(path) 

151 assert clean_path.startswith("/") 

152 return clean_path[1:].replace("/", os.path.sep) 

153 

154 

155def is_path(path): 

156 return os.path.sep in path or (os.path.altsep and os.path.altsep in path) 

157 

158 

159def magic_split_ext(filename, ext_check=True): 

160 """Splits a filename into base and extension. If ext check is enabled 

161 (which is the default) then it verifies the extension is at least 

162 reasonable. 

163 """ 

164 

165 def bad_ext(ext): 

166 if not ext_check: 

167 return False 

168 if not ext or ext.split() != [ext] or ext.strip() != ext: 

169 return True 

170 return False 

171 

172 parts = filename.rsplit(".", 2) 

173 if len(parts) == 1: 

174 return parts[0], "" 

175 if len(parts) == 2 and not parts[0]: 

176 return "." + parts[1], "" 

177 if len(parts) == 3 and len(parts[1]) < 5: 

178 ext = ".".join(parts[1:]) 

179 if not bad_ext(ext): 

180 return parts[0], ext 

181 ext = parts[-1] 

182 if bad_ext(ext): 

183 return filename, "" 

184 basename = ".".join(parts[:-1]) 

185 return basename, ext 

186 

187 

188def iter_dotted_path_prefixes(dotted_path): 

189 pieces = dotted_path.split(".") 

190 if len(pieces) == 1: 

191 yield dotted_path, None 

192 else: 

193 for x in range(1, len(pieces)): 

194 yield ".".join(pieces[:x]), ".".join(pieces[x:]) 

195 

196 

197def resolve_dotted_value(obj, dotted_path): 

198 node = obj 

199 for key in dotted_path.split("."): 

200 if isinstance(node, dict): 

201 new_node = node.get(key) 

202 if new_node is None and key.isdigit(): 

203 new_node = node.get(int(key)) 

204 elif isinstance(node, list): 

205 try: 

206 new_node = node[int(key)] 

207 except (ValueError, TypeError, IndexError): 

208 new_node = None 

209 else: 

210 new_node = None 

211 node = new_node 

212 if node is None: 

213 break 

214 return node 

215 

216 

217def decode_flat_data(itemiter, dict_cls=dict): 

218 def _split_key(name): 

219 result = name.split(".") 

220 for idx, part in enumerate(result): 

221 if part.isdigit(): 

222 result[idx] = int(part) 

223 return result 

224 

225 def _enter_container(container, key): 

226 if key not in container: 

227 return container.setdefault(key, dict_cls()) 

228 return container[key] 

229 

230 def _convert(container): 

231 if _value_marker in container: 

232 force_list = False 

233 values = container.pop(_value_marker) 

234 if container.pop(_list_marker, False): 

235 force_list = True 

236 values.extend(_convert(x[1]) for x in sorted(container.items())) 

237 if not force_list and len(values) == 1: 

238 values = values[0] 

239 

240 if not container: 

241 return values 

242 return _convert(container) 

243 if container.pop(_list_marker, False): 

244 return [_convert(x[1]) for x in sorted(container.items())] 

245 return dict_cls((k, _convert(v)) for k, v in container.items()) 

246 

247 result = dict_cls() 

248 

249 for key, value in itemiter: 

250 parts = _split_key(key) 

251 if not parts: 

252 continue 

253 container = result 

254 for part in parts: 

255 last_container = container 

256 container = _enter_container(container, part) 

257 last_container[_list_marker] = isinstance(part, int) 

258 container[_value_marker] = [value] 

259 

260 return _convert(result) 

261 

262 

263def merge(a, b): 

264 """Merges two values together.""" 

265 if b is None and a is not None: 

266 return a 

267 if a is None: 

268 return b 

269 if isinstance(a, list) and isinstance(b, list): 

270 for idx, (item_1, item_2) in enumerate(zip(a, b, strict=False)): 

271 a[idx] = merge(item_1, item_2) 

272 if isinstance(a, dict) and isinstance(b, dict): 

273 for key, value in b.items(): 

274 a[key] = merge(a.get(key), value) 

275 return a 

276 return a 

277 

278 

279def slugify(text): 

280 """ 

281 A wrapper around python-slugify which preserves file extensions 

282 and forward slashes. 

283 """ 

284 

285 parts = text.split("/") 

286 parts[-1], ext = magic_split_ext(parts[-1]) 

287 

288 out = "/".join(_slugify(part) for part in parts) 

289 

290 if ext: 

291 return out + "." + ext 

292 return out 

293 

294 

295def secure_filename(filename, fallback_name="file"): 

296 base = filename.replace("/", " ").replace("\\", " ") 

297 basename, ext = magic_split_ext(base) 

298 rv = slugify(basename).lstrip(".") 

299 if not rv: 

300 rv = fallback_name 

301 if ext: 

302 return rv + "." + ext 

303 return rv 

304 

305 

306def increment_filename(filename): 

307 directory, filename = os.path.split(filename) 

308 basename, ext = magic_split_ext(filename, ext_check=False) 

309 

310 match = _last_num_re.match(basename) 

311 if match is not None: 

312 rv = match.group(1) + str(int(match.group(2)) + 1) + match.group(3) 

313 else: 

314 rv = basename + "2" 

315 

316 if ext: 

317 rv += "." + ext 

318 if directory: 

319 return os.path.join(directory, rv) 

320 return rv 

321 

322 

323@cache 

324def locate_executable(exe_file, cwd=None, include_bundle_path=True): 

325 """Locates an executable in the search path.""" 

326 choices = [exe_file] 

327 resolve = True 

328 

329 # If it's already a path, we don't resolve. 

330 if os.path.sep in exe_file or (os.path.altsep and os.path.altsep in exe_file): 

331 resolve = False 

332 

333 extensions = os.environ.get("PATHEXT", "").split(";") 

334 _, ext = os.path.splitext(exe_file) 

335 if os.name != "nt" and "" not in extensions or any(ext.lower() == extension.lower() for extension in extensions): 

336 extensions.insert(0, "") 

337 

338 if resolve: 

339 paths = os.environ.get("PATH", "").split(os.pathsep) 

340 choices = [os.path.join(path, exe_file) for path in paths] 

341 

342 if os.name == "nt": 

343 choices.append(os.path.join((cwd or os.getcwd()), exe_file)) 

344 

345 try: 

346 for path in choices: 

347 for ext in extensions: 

348 if os.access(path + ext, os.X_OK): 

349 return path + ext 

350 return None 

351 except OSError: 

352 return None 

353 

354 

355class JSONEncoder(json.JSONEncoder): 

356 def default(self, o): # pylint: disable=method-hidden 

357 if is_undefined(o): 

358 return None 

359 if isinstance(o, datetime): 

360 return http_date(o) 

361 if isinstance(o, uuid.UUID): 

362 return str(o) 

363 if hasattr(o, "__html__"): 

364 return str(o.__html__()) 

365 return json.JSONEncoder.default(self, o) 

366 

367 

368def htmlsafe_json_dump(obj, **kwargs): 

369 kwargs.setdefault("cls", JSONEncoder) 

370 rv = ( 

371 json.dumps(obj, **kwargs) 

372 .replace("<", "\\u003c") 

373 .replace(">", "\\u003e") 

374 .replace("&", "\\u0026") 

375 .replace("'", "\\u0027") 

376 ) 

377 if not _slash_escape: 

378 rv = rv.replace("\\/", "/") 

379 return rv 

380 

381 

382def tojson_filter(obj, **kwargs): 

383 return Markup(htmlsafe_json_dump(obj, **kwargs)) 

384 

385 

386class Url(urllib.parse.SplitResult): 

387 """Make various parts of a URL accessible. 

388 

389 This is the type of the values exposed by Lektor record fields of type "url". 

390 

391 Since Lektor 3.4.0, this is essentially a `urllib.parse.SplitResult` as obtained by 

392 calling `urlsplit` on the URL normalized to an IRI. 

393 

394 Generally, attributes such as ``netloc``, ``host``, ``path``, ``query``, and 

395 ``fragment`` return the IRI (internationalied) versions of those components. 

396 

397 The URI (ASCII-encoded) version of the URL is available from the `ascii_url` 

398 attribute. 

399 

400 NB: Changed in 3.4.0: The ``query`` attribute used to return the URI 

401 (ASCII-encoded) version of the query — I'm not sure why. Now it returns 

402 the IRI (internationalized) version of the query. 

403 

404 """ 

405 

406 url: str 

407 

408 def __new__(cls, value: str): 

409 # XXX: deprecate use of constructor so that eventually we can make its signature 

410 # match that of the SplitResult base class. 

411 warnings.warn( 

412 DeprecatedWarning( 

413 "Url", 

414 reason=( 

415 "Direct construction of a Url instance is deprecated. Use the Url.from_string classmethod instead." 

416 ), 

417 version="3.4.0", 

418 ), 

419 stacklevel=2, 

420 ) 

421 return cls.from_string(value) 

422 

423 @classmethod 

424 def from_string(cls, value: str) -> Url: 

425 """Construct instance from URL string. 

426 

427 The input URL can be a URI (all ASCII) or an IRI (internationalized). 

428 """ 

429 # The iri_to_uri operation is nominally idempotent — it can be passed either an 

430 # IRI or a URI (or something inbetween) and will return a URI. So to fully 

431 # normalize input which can be either an IRI or a URI, first convert to URI, 

432 # then to IRI. 

433 iri = uri_to_iri(iri_to_uri(value)) 

434 obj = cls._make(urllib.parse.urlsplit(iri)) 

435 obj.url = value 

436 return obj 

437 

438 def __str__(self) -> str: 

439 """The original un-normalized URL string.""" 

440 return self.url 

441 

442 @property 

443 def ascii_url(self) -> str: 

444 """The URL encoded to an all-ASCII URI.""" 

445 return iri_to_uri(self.geturl()) 

446 

447 @property 

448 def ascii_host(self) -> str | None: 

449 """The hostname part of the URL IDNA-encoded to ASCII.""" 

450 return urllib.parse.urlsplit(self.ascii_url).hostname 

451 

452 @property 

453 def host(self) -> str | None: 

454 """The IRI (internationalized) version of the hostname. 

455 

456 This attribute is provided for backwards-compatibility. New code should use the 

457 ``hostname`` attribute instead. 

458 """ 

459 return self.hostname 

460 

461 @property 

462 def anchor(self) -> str: 

463 """The IRI (internationalized) version of the "anchor" part of the URL. 

464 

465 This attribute is provided for backwards-compatibility. New code should use the 

466 ``fragment`` attribute instead. 

467 """ 

468 return self.fragment 

469 

470 

471def is_unsafe_to_delete(path, base): 

472 a = os.path.abspath(path) 

473 b = os.path.abspath(base) 

474 diff = os.path.relpath(a, b) 

475 first = diff.split(os.path.sep, maxsplit=1)[0] 

476 return first in (os.path.curdir, os.path.pardir) 

477 

478 

479def prune_file_and_folder(name, base): 

480 if is_unsafe_to_delete(name, base): 

481 return False 

482 try: 

483 os.remove(name) 

484 except OSError: 

485 try: 

486 os.rmdir(name) 

487 except OSError: 

488 return False 

489 head, tail = os.path.split(name) 

490 if not tail: 

491 head, tail = os.path.split(head) 

492 while head and tail: 

493 try: 

494 if is_unsafe_to_delete(head, base): 

495 return False 

496 os.rmdir(head) 

497 except OSError: 

498 break 

499 head, tail = os.path.split(head) 

500 return True 

501 

502 

503def sort_normalize_string(s): 

504 return unicodedata.normalize("NFD", str(s).lower().strip()) 

505 

506 

507def get_dependent_url(url_path, suffix, ext=None): 

508 url_directory, url_filename = posixpath.split(url_path) 

509 url_base, url_ext = posixpath.splitext(url_filename) 

510 if ext is None: 

511 ext = url_ext 

512 return posixpath.join(url_directory, url_base + "@" + suffix + ext) 

513 

514 

515# These are the only modes we really support 

516_AtomicOpenTextMode = Literal["w", "wt", "tw", "r", "rt", "tr"] 

517_AtomicOpenBinaryModeWriting = Literal["wb", "bw"] 

518_AtomicOpenBinaryModeReading = Literal["rb", "br"] 

519_AtomicOpenMode = _AtomicOpenTextMode | _AtomicOpenBinaryModeWriting | _AtomicOpenBinaryModeReading 

520 

521 

522@overload 

523@contextmanager 

524def atomic_open( 

525 filename: StrPath, 

526 mode: _AtomicOpenTextMode = "r", 

527 encoding: str | None = None, 

528) -> Iterator[io.TextIOWrapper]: ... 

529 

530 

531@overload 

532@contextmanager 

533def atomic_open( 

534 filename: StrPath, 

535 mode: _AtomicOpenBinaryModeWriting, 

536 encoding: None = None, 

537) -> Iterator[io.BufferedWriter]: ... 

538 

539 

540@overload 

541@contextmanager 

542def atomic_open( 

543 filename: StrPath, 

544 mode: _AtomicOpenBinaryModeReading, 

545 encoding: None = None, 

546) -> Iterator[io.BufferedReader]: ... 

547 

548 

549@contextmanager 

550def atomic_open(filename: StrPath, mode: _AtomicOpenMode = "r", encoding: str | None = None) -> Iterator[IO[Any]]: 

551 """Open a file for atomic update. 

552 

553 Perform an "all-or-nothing" write to a file. 

554 

555 This opens a temporary file to receive writes. It is meant to be used as a context 

556 manager. When the context is exited normally, the temporary file is closed then 

557 atomically renamed to the target file name. 

558 

559 If an exception is thrown during this process the temporary file is silently 

560 deleted. 

561 

562 """ 

563 if any(c in mode for c in "ax+"): 

564 raise ValueError(f"unsupported open mode: {mode}") 

565 

566 if "r" in mode: 

567 with open(filename, mode=mode, encoding=encoding) as fp: 

568 yield fp 

569 return 

570 

571 fd, tmp_filename = create_temp( 

572 prefix=".__atomic-write", 

573 dir=Path(filename).parent, 

574 text="b" not in mode, 

575 ) 

576 try: 

577 with open(fd, mode, encoding=encoding) as fp: 

578 yield fp 

579 os.replace(tmp_filename, filename) 

580 except Exception: 

581 with suppress(OSError): 

582 os.remove(tmp_filename) 

583 raise 

584 

585 

586def create_temp( 

587 suffix: str = "", 

588 prefix: str = "tmp", 

589 dir: StrPath | None = None, 

590 text: bool = False, 

591 mode: int = 0o666, 

592) -> tuple[int, str | bytes]: 

593 """Create and return a unique temporary file. 

594 

595 The return value is a pair (fd, name) where fd is the file descriptor returned by 

596 os.open, and name is the filename. 

597 

598 This works very much like `tempfile.mkstemp`, except that it allows more control 

599 over the mode (access permissions) of the created file. 

600 

601 The access permissions of the created file is determined by the value of 'mode' 

602 (which defaults to 0o666) combined with any umask or default ACL that may be in 

603 place. 

604 

605 """ 

606 if (tmp_dir := dir) is None: 

607 tmp_dir = os.getcwd() 

608 

609 flags = os.O_RDWR | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) 

610 if not text: 

611 flags |= getattr(os, "O_BINARY", 0) 

612 

613 for _ in range(32000): 

614 filename = tempfile.mktemp(suffix=suffix, prefix=prefix, dir=tmp_dir) 

615 try: 

616 fd = os.open(filename, flags, mode) 

617 except FileExistsError: 

618 continue 

619 # NB: Under Windows, PermissionError can be raised by os.open if a directory 

620 # with the chosen name already exists. Since mktemp doesn't return names to 

621 # existing files/dirs, the likelihood of this happening is slim. For now, we 

622 # don't check for it here. 

623 return fd, filename 

624 

625 raise AssertionError("Unable to find temporary file name") 

626 

627 

628def portable_popen(cmd, *args, **kwargs): 

629 """A portable version of subprocess.Popen that automatically locates 

630 executables before invoking them. This also looks for executables 

631 in the bundle bin. 

632 """ 

633 if cmd[0] is None: 

634 raise RuntimeError("No executable specified") 

635 exe = locate_executable(cmd[0], kwargs.get("cwd")) 

636 if exe is None: 

637 raise RuntimeError(f'Could not locate executable "{cmd[0]}"') 

638 

639 if isinstance(exe, str) and sys.platform != "win32": 

640 exe = exe.encode(sys.getfilesystemencoding()) 

641 cmd[0] = exe 

642 return subprocess.Popen(cmd, *args, **kwargs) 

643 

644 

645def is_valid_id(value): 

646 if value == "": 

647 return True 

648 return "/" not in value and value.strip() == value and value.split() == [value] and not value.startswith(".") 

649 

650 

651def secure_url(url: str) -> str: 

652 parts = urllib.parse.urlsplit(url) 

653 if parts.password is not None: 

654 _, _, host_port = parts.netloc.rpartition("@") 

655 parts = parts._replace(netloc=f"{parts.username}@{host_port}") 

656 return parts.geturl() 

657 

658 

659def bool_from_string(val, default=None): 

660 if val in (True, False, 1, 0): 

661 return bool(val) 

662 if isinstance(val, str): 

663 val = val.lower() 

664 if val in ("true", "yes", "1"): 

665 return True 

666 if val in ("false", "no", "0"): 

667 return False 

668 return default 

669 

670 

671def make_relative_url(source, target): 

672 """ 

673 Returns the relative path (url) needed to navigate 

674 from `source` to `target`. 

675 """ 

676 

677 # WARNING: this logic makes some unwarranted assumptions about 

678 # what is a directory and what isn't. Ideally, this function 

679 # would be aware of the actual filesystem. 

680 s_is_dir = source.endswith("/") 

681 t_is_dir = target.endswith("/") 

682 

683 source = PurePosixPath(posixpath.normpath(source)) 

684 target = PurePosixPath(posixpath.normpath(target)) 

685 

686 if not s_is_dir: 

687 source = source.parent 

688 

689 relpath = str(get_relative_path(source, target)) 

690 if t_is_dir: 

691 relpath += "/" 

692 

693 return relpath 

694 

695 

696def get_relative_path(source, target): 

697 """ 

698 Returns the relative path needed to navigate from `source` to `target`. 

699 

700 get_relative_path(source: PurePosixPath, 

701 target: PurePosixPath) -> PurePosixPath 

702 """ 

703 

704 if not source.is_absolute() and target.is_absolute(): 

705 raise ValueError("Cannot navigate from a relative path to an absolute one") 

706 

707 if source.is_absolute() and not target.is_absolute(): 

708 # nothing to do 

709 return target 

710 

711 if source.is_absolute() and target.is_absolute(): 

712 # convert them to relative paths to simplify the logic 

713 source = source.relative_to("/") 

714 target = target.relative_to("/") 

715 

716 # is the source an ancestor of the target? 

717 try: 

718 return target.relative_to(source) 

719 except ValueError: 

720 pass 

721 

722 # even if it isn't, one of the source's ancestors might be 

723 # (and if not, the root will be the common ancestor) 

724 distance = PurePosixPath(".") 

725 for ancestor in source.parents: 

726 distance /= ".." 

727 

728 try: 

729 relpath = target.relative_to(ancestor) 

730 except ValueError: 

731 continue 

732 else: 

733 # prepend the distance to the common ancestor 

734 return distance / relpath 

735 # We should never get here. (The last ancestor in source.parents will 

736 # be '.' — target.relative_to('.') will always succeed.) 

737 raise AssertionError("This should not happen") 

738 

739 

740def deg_to_dms(deg): 

741 d = int(deg) 

742 md = abs(deg - d) * 60 

743 m = int(md) 

744 sd = (md - m) * 60 

745 return (d, m, sd) 

746 

747 

748def format_lat_long(lat=None, long=None, secs=True): 

749 def _format(value, sign): 

750 d, m, sd = deg_to_dms(value) 

751 return f"{abs(d)}° {abs(m)}{secs and f'{abs(sd)}' or ''}{sign[d < 0]}" 

752 

753 rv = [] 

754 if lat is not None: 

755 rv.append(_format(lat, "NS")) 

756 if long is not None: 

757 rv.append(_format(long, "EW")) 

758 return ", ".join(rv) 

759 

760 

761def split_camel_case(s: str) -> list[str]: 

762 """Split camel-cased words. 

763 

764 This currently only works with ASCII strings. 

765 """ 

766 return re.split(r"(?<=[a-z])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])|\s+", s.strip()) 

767 

768 

769def get_cache_dir(): 

770 if is_windows: 

771 folder = os.environ.get("LOCALAPPDATA") 

772 if folder is None: 

773 folder = os.environ.get("APPDATA") 

774 if folder is None: 

775 folder = os.path.expanduser("~") 

776 return os.path.join(folder, "Lektor", "Cache") 

777 if sys.platform == "darwin": 

778 return os.path.join(os.path.expanduser("~/Library/Caches/Lektor")) 

779 return os.path.join(os.environ.get("XDG_CACHE_HOME", os.path.expanduser("~/.cache")), "lektor") 

780 

781 

782class URLBuilder: 

783 def __init__(self): 

784 self.items = [] 

785 

786 def append(self, item): 

787 if item is None: 

788 return 

789 item = str(item).strip("/") 

790 if item: 

791 self.items.append(item) 

792 

793 def get_url(self, trailing_slash=None): 

794 url = "/" + "/".join(self.items) 

795 if trailing_slash is not None and not trailing_slash: 

796 return url 

797 if url == "/": 

798 return url 

799 if trailing_slash is None: 

800 _, last = url.rsplit("/", 1) 

801 if "." in last: 

802 # Assuming the url points to a file 

803 return url 

804 return url + "/" 

805 

806 

807def build_url(iterable, trailing_slash=None): 

808 # NB: While this function is not used by Lektor itself, it is used 

809 # by a number of plugins including: lektor-atom, 

810 # lektor-gemini-capsule, lektor-index-pages, and lektor-tags. 

811 builder = URLBuilder() 

812 for item in iterable: 

813 builder.append(item) 

814 return builder.get_url(trailing_slash=trailing_slash) 

815 

816 

817def comma_delimited(s): 

818 """Split a comma-delimited string.""" 

819 for part in s.split(","): 

820 stripped = part.strip() 

821 if stripped: 

822 yield stripped 

823 

824 

825def process_extra_flags(flags): 

826 if isinstance(flags, dict): 

827 return flags 

828 rv = {} 

829 for flag in flags or (): 

830 if ":" in flag: 

831 k, v = flag.split(":", 1) 

832 rv[k] = v 

833 else: 

834 rv[flag] = flag 

835 return rv 

836 

837 

838_H = TypeVar("_H", bound=Hashable) 

839 

840 

841def unique_everseen(seq: Iterable[_H]) -> Iterable[_H]: 

842 """Filter out duplicates from iterable.""" 

843 # This is a less general version of more_itertools.unique_everseen. 

844 # Should we need more general functionality, consider using that instead. 

845 seen = set() 

846 for val in seq: 

847 if val not in seen: 

848 seen.add(val) 

849 yield val 

850 

851 

852class RecursionCheck(threading.local): 

853 """A context manager that retains a count of how many times it's been entered. 

854 

855 Example: 

856 

857 >>> recursion_check = RecursionCheck() 

858 

859 >>> with recursion_check: 

860 ... assert recursion_check.level == 1 

861 ... with recursion_check as recursion_level: 

862 ... assert recursion_check.level == 2 

863 ... print("depth", recursion_level) 

864 ... assert recursion_check.level == 1 

865 ... assert recursion_check.level == 0 

866 depth 2 

867 """ 

868 

869 level = 0 

870 

871 def __enter__(self) -> int: 

872 self.level += 1 

873 return self.level 

874 

875 def __exit__(self, _t, _v, _tb) -> None: 

876 self.level -= 1 

877 

878 

879class DeprecatedWarning(DeprecationWarning): 

880 """Warning category issued by our ``deprecated`` decorator.""" 

881 

882 def __init__( 

883 self, 

884 name: str, 

885 reason: str | None = None, 

886 version: str | None = None, 

887 ): 

888 self.name = name 

889 self.reason = reason 

890 self.version = version 

891 

892 def __str__(self) -> str: 

893 message = f"{self.name!r} is deprecated" 

894 if self.reason: 

895 message += f" ({self.reason})" 

896 if self.version: 

897 message += f" since version {self.version}" 

898 return message 

899 

900 

901_F = TypeVar("_F", bound=Callable[..., Any]) 

902 

903 

904@dataclass 

905class _Deprecate: 

906 """A decorator to mark callables as deprecated.""" 

907 

908 name: str | None = None 

909 reason: str | None = None 

910 version: str | None = None 

911 stacklevel: int = 1 

912 

913 _recursion_check: ClassVar = RecursionCheck() 

914 

915 def __call__(self, wrapped: _F) -> _F: 

916 if not callable(wrapped): 

917 raise TypeError("do not know how to deprecate {wrapped!r}") 

918 

919 name = self.name or wrapped.__name__ 

920 message = DeprecatedWarning(name, self.reason, self.version) 

921 

922 @wraps(wrapped) 

923 def wrapper(*args: Any, **kwargs: Any) -> Any: 

924 with self._recursion_check as recursion_level: 

925 if recursion_level == 1: 

926 warnings.warn(message, stacklevel=self.stacklevel + 1) 

927 return wrapped(*args, **kwargs) 

928 

929 return wrapper # type: ignore[return-value] 

930 

931 

932@overload 

933def deprecated( 

934 __wrapped: Callable[..., Any], 

935 *, 

936 name: str | None = ..., 

937 reason: str | None = ..., 

938 version: str | None = ..., 

939 stacklevel: int = ..., 

940) -> Callable[..., Any]: ... 

941 

942 

943@overload 

944def deprecated( 

945 __reason: str, 

946 *, 

947 name: str | None = ..., 

948 version: str | None = ..., 

949 stacklevel: int = ..., 

950) -> _Deprecate: ... 

951 

952 

953@overload 

954def deprecated( 

955 *, 

956 name: str | None = ..., 

957 reason: str | None = ..., 

958 version: str | None = ..., 

959 stacklevel: int = ..., 

960) -> _Deprecate: ... 

961 

962 

963def deprecated(*args: Any, **kwargs: Any) -> _F | _Deprecate: 

964 """A decorator to mark callables or descriptors as deprecated. 

965 

966 This can be used to decorate functions, methods, classes, and descriptors. 

967 In particular, this decorator can be applied to instances of ``property``, 

968 ``functools.cached_property`` and ``werkzeug.utils.cached_property``. 

969 

970 When the decorated object is called (or — in the case of a descriptor — accessed), a 

971 ``DeprecationWarning`` is issued. 

972 

973 The warning message will include the name of the decorated object, and may include 

974 further information if provided from the ``reason`` and ``version`` arguments. 

975 

976 The ``name`` argument may be used to specify an alternative name to use when 

977 generating the warning message. By default, the ``__name__`` attribute of the 

978 decorated object is used. 

979 

980 The ``stacklevel`` argument controls which call in the call stack the warning 

981 is attributed to. The default value, ``stacklevel=1`` means the warning is 

982 reported for the immediate caller of the decorated object. Higher values 

983 attribute the warning callers further back in the stack. 

984 

985 """ 

986 if len(args) > 1: 

987 raise TypeError("deprecated accepts a maximum of one positional parameter") 

988 

989 wrapped: _F | None = None 

990 if args: 

991 if isinstance(args[0], str): 

992 kwargs.setdefault("reason", args[0]) 

993 else: 

994 wrapped = args[0] 

995 

996 deprecate = _Deprecate(**kwargs) 

997 if wrapped is not None: 

998 return deprecate(wrapped) 

999 return deprecate