Coverage for src/lektor_ng/utils.py: 81%
545 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-03 22:08 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-03 22:08 +0000
1from __future__ import annotations
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
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
31if TYPE_CHECKING:
32 from _typeshed import StrPath
35is_windows = os.name == "nt"
37_slash_escape = "\\/" not in json.dumps("/")
39_last_num_re = re.compile(r"^(.*)(\d+)(.*?)$")
40_list_marker = object()
41_value_marker = object()
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
52def split_virtual_path(path):
53 if "@" in path:
54 return path.split("@", 1)
55 return path, None
58def _norm_join(a, b):
59 return posixpath.normpath(posixpath.join(a, b))
62def join_path(a, b):
63 """Join two DB-paths.
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.
72 if posixpath.isabs(b):
73 return b
75 a_p, a_v = split_virtual_path(a)
76 b_p, b_v = split_virtual_path(b)
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
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
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("/"))
103def cleanup_url_path(url_path):
104 """Clean up a URL path.
106 This strips any query, and/or fragment that may be present in the
107 input path.
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}")
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("/"))
123def parse_path(path):
124 x = cleanup_path(path).strip("/").split("/")
125 if x == [""]:
126 return []
127 return x
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 "")
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)
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)
155def is_path(path):
156 return os.path.sep in path or (os.path.altsep and os.path.altsep in path)
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 """
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
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
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:])
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
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
225 def _enter_container(container, key):
226 if key not in container:
227 return container.setdefault(key, dict_cls())
228 return container[key]
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]
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())
247 result = dict_cls()
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]
260 return _convert(result)
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
279def slugify(text):
280 """
281 A wrapper around python-slugify which preserves file extensions
282 and forward slashes.
283 """
285 parts = text.split("/")
286 parts[-1], ext = magic_split_ext(parts[-1])
288 out = "/".join(_slugify(part) for part in parts)
290 if ext:
291 return out + "." + ext
292 return out
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
306def increment_filename(filename):
307 directory, filename = os.path.split(filename)
308 basename, ext = magic_split_ext(filename, ext_check=False)
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"
316 if ext:
317 rv += "." + ext
318 if directory:
319 return os.path.join(directory, rv)
320 return rv
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
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
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, "")
338 if resolve:
339 paths = os.environ.get("PATH", "").split(os.pathsep)
340 choices = [os.path.join(path, exe_file) for path in paths]
342 if os.name == "nt":
343 choices.append(os.path.join((cwd or os.getcwd()), exe_file))
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
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)
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
382def tojson_filter(obj, **kwargs):
383 return Markup(htmlsafe_json_dump(obj, **kwargs))
386class Url(urllib.parse.SplitResult):
387 """Make various parts of a URL accessible.
389 This is the type of the values exposed by Lektor record fields of type "url".
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.
394 Generally, attributes such as ``netloc``, ``host``, ``path``, ``query``, and
395 ``fragment`` return the IRI (internationalied) versions of those components.
397 The URI (ASCII-encoded) version of the URL is available from the `ascii_url`
398 attribute.
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.
404 """
406 url: str
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)
423 @classmethod
424 def from_string(cls, value: str) -> Url:
425 """Construct instance from URL string.
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
438 def __str__(self) -> str:
439 """The original un-normalized URL string."""
440 return self.url
442 @property
443 def ascii_url(self) -> str:
444 """The URL encoded to an all-ASCII URI."""
445 return iri_to_uri(self.geturl())
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
452 @property
453 def host(self) -> str | None:
454 """The IRI (internationalized) version of the hostname.
456 This attribute is provided for backwards-compatibility. New code should use the
457 ``hostname`` attribute instead.
458 """
459 return self.hostname
461 @property
462 def anchor(self) -> str:
463 """The IRI (internationalized) version of the "anchor" part of the URL.
465 This attribute is provided for backwards-compatibility. New code should use the
466 ``fragment`` attribute instead.
467 """
468 return self.fragment
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)
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
503def sort_normalize_string(s):
504 return unicodedata.normalize("NFD", str(s).lower().strip())
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)
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
522@overload
523@contextmanager
524def atomic_open(
525 filename: StrPath,
526 mode: _AtomicOpenTextMode = "r",
527 encoding: str | None = None,
528) -> Iterator[io.TextIOWrapper]: ...
531@overload
532@contextmanager
533def atomic_open(
534 filename: StrPath,
535 mode: _AtomicOpenBinaryModeWriting,
536 encoding: None = None,
537) -> Iterator[io.BufferedWriter]: ...
540@overload
541@contextmanager
542def atomic_open(
543 filename: StrPath,
544 mode: _AtomicOpenBinaryModeReading,
545 encoding: None = None,
546) -> Iterator[io.BufferedReader]: ...
549@contextmanager
550def atomic_open(filename: StrPath, mode: _AtomicOpenMode = "r", encoding: str | None = None) -> Iterator[IO[Any]]:
551 """Open a file for atomic update.
553 Perform an "all-or-nothing" write to a file.
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.
559 If an exception is thrown during this process the temporary file is silently
560 deleted.
562 """
563 if any(c in mode for c in "ax+"):
564 raise ValueError(f"unsupported open mode: {mode}")
566 if "r" in mode:
567 with open(filename, mode=mode, encoding=encoding) as fp:
568 yield fp
569 return
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
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.
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.
598 This works very much like `tempfile.mkstemp`, except that it allows more control
599 over the mode (access permissions) of the created file.
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.
605 """
606 if (tmp_dir := dir) is None:
607 tmp_dir = os.getcwd()
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)
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
625 raise AssertionError("Unable to find temporary file name")
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]}"')
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)
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(".")
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()
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
671def make_relative_url(source, target):
672 """
673 Returns the relative path (url) needed to navigate
674 from `source` to `target`.
675 """
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("/")
683 source = PurePosixPath(posixpath.normpath(source))
684 target = PurePosixPath(posixpath.normpath(target))
686 if not s_is_dir:
687 source = source.parent
689 relpath = str(get_relative_path(source, target))
690 if t_is_dir:
691 relpath += "/"
693 return relpath
696def get_relative_path(source, target):
697 """
698 Returns the relative path needed to navigate from `source` to `target`.
700 get_relative_path(source: PurePosixPath,
701 target: PurePosixPath) -> PurePosixPath
702 """
704 if not source.is_absolute() and target.is_absolute():
705 raise ValueError("Cannot navigate from a relative path to an absolute one")
707 if source.is_absolute() and not target.is_absolute():
708 # nothing to do
709 return target
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("/")
716 # is the source an ancestor of the target?
717 try:
718 return target.relative_to(source)
719 except ValueError:
720 pass
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 /= ".."
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")
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)
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]}"
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)
761def split_camel_case(s: str) -> list[str]:
762 """Split camel-cased words.
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())
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")
782class URLBuilder:
783 def __init__(self):
784 self.items = []
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)
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 + "/"
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)
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
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
838_H = TypeVar("_H", bound=Hashable)
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
852class RecursionCheck(threading.local):
853 """A context manager that retains a count of how many times it's been entered.
855 Example:
857 >>> recursion_check = RecursionCheck()
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 """
869 level = 0
871 def __enter__(self) -> int:
872 self.level += 1
873 return self.level
875 def __exit__(self, _t, _v, _tb) -> None:
876 self.level -= 1
879class DeprecatedWarning(DeprecationWarning):
880 """Warning category issued by our ``deprecated`` decorator."""
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
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
901_F = TypeVar("_F", bound=Callable[..., Any])
904@dataclass
905class _Deprecate:
906 """A decorator to mark callables as deprecated."""
908 name: str | None = None
909 reason: str | None = None
910 version: str | None = None
911 stacklevel: int = 1
913 _recursion_check: ClassVar = RecursionCheck()
915 def __call__(self, wrapped: _F) -> _F:
916 if not callable(wrapped):
917 raise TypeError("do not know how to deprecate {wrapped!r}")
919 name = self.name or wrapped.__name__
920 message = DeprecatedWarning(name, self.reason, self.version)
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)
929 return wrapper # type: ignore[return-value]
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]: ...
943@overload
944def deprecated(
945 __reason: str,
946 *,
947 name: str | None = ...,
948 version: str | None = ...,
949 stacklevel: int = ...,
950) -> _Deprecate: ...
953@overload
954def deprecated(
955 *,
956 name: str | None = ...,
957 reason: str | None = ...,
958 version: str | None = ...,
959 stacklevel: int = ...,
960) -> _Deprecate: ...
963def deprecated(*args: Any, **kwargs: Any) -> _F | _Deprecate:
964 """A decorator to mark callables or descriptors as deprecated.
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``.
970 When the decorated object is called (or — in the case of a descriptor — accessed), a
971 ``DeprecationWarning`` is issued.
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.
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.
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.
985 """
986 if len(args) > 1:
987 raise TypeError("deprecated accepts a maximum of one positional parameter")
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]
996 deprecate = _Deprecate(**kwargs)
997 if wrapped is not None:
998 return deprecate(wrapped)
999 return deprecate