Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/util/langhelpers.py: 51%

1# util/langhelpers.py

3# <see AUTHORS file>

5# This module is part of SQLAlchemy and is released under

6# the MIT License: https://www.opensource.org/licenses/mit-license.php

7# mypy: allow-untyped-defs, allow-untyped-calls

9"""Routines to help with the creation, loading and introspection of

10modules, classes, hierarchies, attributes, functions, and methods.

12"""

14from __future__ import annotations

16import collections

17import enum

18from functools import update_wrapper

19import importlib.metadata

20import importlib.util

21import inspect

22import itertools

23import linecache

24import operator

25import re

26import sys

27import textwrap

28import threading

29import types

30from types import CodeType

31from types import ModuleType

32from typing import Any

33from typing import Callable

34from typing import cast

35from typing import Dict

36from typing import FrozenSet

37from typing import Generic

38from typing import Iterator

39from typing import List

40from typing import Literal

41from typing import NoReturn

42from typing import Optional

43from typing import overload

44from typing import Sequence

45from typing import Set

46from typing import Tuple

47from typing import Type

48from typing import TYPE_CHECKING

49from typing import TypeVar

50from typing import Union

51import warnings

52import weakref

54from . import _collections

55from . import compat

56from .. import exc

58_T = TypeVar("_T")

59_T_co = TypeVar("_T_co", covariant=True)

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

61_MA = TypeVar("_MA", bound="HasMemoized.memoized_attribute[Any]")

62_M = TypeVar("_M", bound=ModuleType)

65def restore_annotations(

66 cls: type, new_annotations: dict[str, Any]

67) -> Callable[[], None]:

68 """apply alternate annotations to a class, with a callable to restore

69 the pristine state of the former.

70 This is used strictly to provide dataclasses on a mapped class, where

71 in some cases where are making dataclass fields based on an attribute

72 that is actually a python descriptor on a superclass which we called

73 to get a value.

74 if dataclasses were to give us a way to achieve this without swapping

75 __annotations__, that would be much better.

76 """

77 delattr_ = object()

79 # pep-649 means classes have "__annotate__", and it's a callable. if it's

80 # there and is None, we're in "legacy future mode", where it's python 3.14

81 # or higher and "from __future__ import annotations" is set. in "legacy

82 # future mode" we have to do the same steps we do for older pythons,

83 # __annotate__ can be ignored

84 is_pep649 = hasattr(cls, "__annotate__") and cls.__annotate__ is not None

86 if is_pep649:

87 memoized = {

88 "__annotate__": getattr(cls, "__annotate__", delattr_),

89 }

90 else:

91 memoized = {

92 "__annotations__": getattr(cls, "__annotations__", delattr_)

93 }

95 cls.__annotations__ = new_annotations

97 def restore():

98 for k, v in memoized.items():

99 if v is delattr_:

100 delattr(cls, k)

101 else:

102 setattr(cls, k, v)

103

104 return restore

105

106

107def md5_hex(x: Any) -> str:

108 x = x.encode("utf-8")

109 m = compat.md5_not_for_security()

110 m.update(x)

111 return cast(str, m.hexdigest())

112

113

114class safe_reraise:

115 """Reraise an exception after invoking some

116 handler code.

117

118 Stores the existing exception info before

119 invoking so that it is maintained across a potential

120 coroutine context switch.

121

122 e.g.::

123

124 try:

125 sess.commit()

126 except:

127 with safe_reraise():

128 sess.rollback()

129

130 TODO: we should at some point evaluate current behaviors in this regard

131 based on current greenlet, gevent/eventlet implementations in Python 3, and

132 also see the degree to which our own asyncio (based on greenlet also) is

133 impacted by this. .rollback() will cause IO / context switch to occur in

134 all these scenarios; what happens to the exception context from an

135 "except:" block if we don't explicitly store it? Original issue was #2703.

136

137 """

138

139 __slots__ = ("_exc_info",)

140

141 _exc_info: Union[

142 None,

143 Tuple[

144 Type[BaseException],

145 BaseException,

146 types.TracebackType,

147 ],

148 Tuple[None, None, None],

149 ]

150

151 def __enter__(self) -> None:

152 self._exc_info = sys.exc_info()

153

154 def __exit__(

155 self,

156 type_: Optional[Type[BaseException]],

157 value: Optional[BaseException],

158 traceback: Optional[types.TracebackType],

159 ) -> NoReturn:

160 assert self._exc_info is not None

161 # see #2703 for notes

162 if type_ is None:

163 exc_type, exc_value, exc_tb = self._exc_info

164 assert exc_value is not None

165 self._exc_info = None # remove potential circular references

166 raise exc_value.with_traceback(exc_tb)

167 else:

168 self._exc_info = None # remove potential circular references

169 assert value is not None

170 raise value.with_traceback(traceback)

171

172

173def walk_subclasses(cls: Type[_T]) -> Iterator[Type[_T]]:

174 seen: Set[Any] = set()

175

176 stack = [cls]

177 while stack:

178 cls = stack.pop()

179 if cls in seen:

180 continue

181 else:

182 seen.add(cls)

183 stack.extend(cls.__subclasses__())

184 yield cls

185

186

187def string_or_unprintable(element: Any) -> str:

188 if isinstance(element, str):

189 return element

190 else:

191 try:

192 return str(element)

193 except Exception:

194 return "unprintable element %r" % element

195

196

197def clsname_as_plain_name(

198 cls: Type[Any], use_name: Optional[str] = None

199) -> str:

200 name = use_name or cls.__name__

201 return " ".join(n.lower() for n in re.findall(r"([A-Z][a-z]+|SQL)", name))

202

203

204def method_is_overridden(

205 instance_or_cls: Union[Type[Any], object],

206 against_method: Callable[..., Any],

207) -> bool:

208 """Return True if the two class methods don't match."""

209

210 if not isinstance(instance_or_cls, type):

211 current_cls = instance_or_cls.__class__

212 else:

213 current_cls = instance_or_cls

214

215 method_name = against_method.__name__

216

217 current_method: types.MethodType = getattr(current_cls, method_name)

218

219 return current_method != against_method

220

221

222def decode_slice(slc: slice) -> Tuple[Any, ...]:

223 """decode a slice object as sent to __getitem__.

224

225 takes into account the 2.5 __index__() method, basically.

226

227 """

228 ret: List[Any] = []

229 for x in slc.start, slc.stop, slc.step:

230 if hasattr(x, "__index__"):

231 x = x.__index__()

232 ret.append(x)

233 return tuple(ret)

234

235

236def _unique_symbols(used: Sequence[str], *bases: str) -> Iterator[str]:

237 used_set = set(used)

238 for base in bases:

239 pool = itertools.chain(

240 (base,),

241 map(lambda i: base + str(i), range(1000)),

242 )

243 for sym in pool:

244 if sym not in used_set:

245 used_set.add(sym)

246 yield sym

247 break

248 else:

249 raise NameError("exhausted namespace for symbol base %s" % base)

250

251

252def map_bits(fn: Callable[[int], Any], n: int) -> Iterator[Any]:

253 """Call the given function given each nonzero bit from n."""

254

255 while n:

256 b = n & (~n + 1)

257 yield fn(b)

258 n ^= b

259

260

261_Fn = TypeVar("_Fn", bound="Callable[..., Any]")

262

263# this seems to be in flux in recent mypy versions

264

265

266def decorator(target: Callable[..., Any]) -> Callable[[_Fn], _Fn]:

267 """A signature-matching decorator factory."""

268

269 def decorate(fn: _Fn) -> _Fn:

270 if not inspect.isfunction(fn) and not inspect.ismethod(fn):

271 raise Exception("not a decoratable function")

272

273 # Python 3.14 defer creating __annotations__ until its used.

274 # We do not want to create __annotations__ now.

275 annofunc = getattr(fn, "__annotate__", None)

276 if annofunc is not None:

277 fn.__annotate__ = None # type: ignore[union-attr]

278 try:

279 spec = compat.inspect_getfullargspec(fn)

280 finally:

281 fn.__annotate__ = annofunc # type: ignore[union-attr]

282 else:

283 spec = compat.inspect_getfullargspec(fn)

284

285 # Do not generate code for annotations.

286 # update_wrapper() copies the annotation from fn to decorated.

287 # We use dummy defaults for code generation to avoid having

288 # copy of large globals for compiling.

289 # We copy __defaults__ and __kwdefaults__ from fn to decorated.

290 empty_defaults = (None,) * len(spec.defaults or ())

291 empty_kwdefaults = dict.fromkeys(spec.kwonlydefaults or ())

292 spec = spec._replace(

293 annotations={},

294 defaults=empty_defaults,

295 kwonlydefaults=empty_kwdefaults,

296 )

297

298 names = (

299 tuple(cast("Tuple[str, ...]", spec[0]))

300 + cast("Tuple[str, ...]", spec[1:3])

301 + (fn.__name__,)

302 )

303 targ_name, fn_name = _unique_symbols(names, "target", "fn")

304

305 metadata: Dict[str, Optional[str]] = dict(target=targ_name, fn=fn_name)

306 metadata.update(format_argspec_plus(spec, grouped=False))

307 metadata["name"] = fn.__name__

308

309 if inspect.iscoroutinefunction(fn):

310 metadata["prefix"] = "async "

311 metadata["target_prefix"] = "await "

312 metadata["target_suffix"] = ""

313 elif inspect.isgeneratorfunction(fn):

314 # a generator function has to remain a generator function

315 # after decoration; tools such as pytest fixtures test for

316 # inspect.isgeneratorfunction() and will otherwise never

317 # iterate the function at all

318 metadata["prefix"] = ""

319 metadata["target_prefix"] = "(yield from "

320 metadata["target_suffix"] = ")"

321 else:

322 metadata["prefix"] = ""

323 metadata["target_prefix"] = ""

324 metadata["target_suffix"] = ""

325

326 # look for __ positional arguments. This is a convention in

327 # SQLAlchemy that arguments should be passed positionally

328 # rather than as keyword

329 # arguments. note that apply_pos doesn't currently work in all cases

330 # such as when a kw-only indicator "*" is present, which is why

331 # we limit the use of this to just that case we can detect. As we add

332 # more kinds of methods that use @decorator, things may have to

333 # be further improved in this area

334 if "__" in repr(spec[0]):

335 code = """\

336%(prefix)sdef %(name)s%(grouped_args)s:

337 return %(target_prefix)s%(target)s(%(fn)s, %(apply_pos)s)%(target_suffix)s

338""" % metadata

339 else:

340 code = """\

341%(prefix)sdef %(name)s%(grouped_args)s:

342 return %(target_prefix)s%(target)s(%(fn)s, %(apply_kw)s)%(target_suffix)s

343""" % metadata

344

345 env: Dict[str, Any] = {

346 targ_name: target,

347 fn_name: fn,

348 "__name__": fn.__module__,

349 }

350

351 # the target's name is part of the description because decorators

352 # built here do get stacked (see ValuesBase.values()), and

353 # update_wrapper() gives every layer the same __qualname__; without

354 # it the outer layer would claim the inner layer's source.

355 decorated = cast(

356 types.FunctionType,

357 exec_code_in_env(

358 code,

359 env,

360 fn.__name__,

361 f"{target.__name__}() wrapper for "

362 f"{fn.__module__}.{fn.__qualname__}",

363 ),

364 )

365 decorated.__defaults__ = fn.__defaults__

366 decorated.__kwdefaults__ = fn.__kwdefaults__ # type: ignore[union-attr] # noqa: E501

367 return update_wrapper(decorated, fn) # type: ignore[return-value]

368

369 return update_wrapper(decorate, target) # type: ignore[return-value]

370

371

372_LinecacheEntry = Tuple[int, None, List[str], str]

373

374

375def _linecache_cache_getter():

376 """safe getter for linecache.cache

377

378 linecache.cache despite being non-underscored and widely used is

379 nonetheless not documented by cPython. Therefore we cannot trust that it

380 it's present in third party Python distributions, or that it wont

381 suddenly be removed or changed. Guard against this such that linecache

382 features will be silently disabled if this should happen. Unit tests in

383 test/base/test_utils.py ensures linecache.cache remains available for new

384 releases.

385

386 """

387 try:

388 linecache_cache = linecache.cache

389 except AttributeError:

390 raise

391 else:

392 if not isinstance(linecache_cache, dict):

393 raise AttributeError("linecache has changed from being a dict")

394 return linecache_cache

395

396

397def _remove_linecache_entry(filename: str, entry: _LinecacheEntry) -> None:

398 """Discard a ``linecache`` entry made by :func:`.exec_code_in_env`, if

399 it is still the live one.

400

401 Runs from a weakref finalizer.

402

403 This function is only established if we actually added an entry to the

404 linecache within exec_code_in_env.

405

406 """

407 linecache_cache = _linecache_cache_getter()

408

409 if linecache_cache.get(filename) is entry:

410 linecache_cache.pop(filename, None)

411

412

413def exec_code_in_env(

414 code: Union[str, types.CodeType],

415 env: Dict[str, Any],

416 fn_name: str,

417 description: Optional[str] = None,

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

419 """Exec generated ``code`` in ``env`` and return the function it defines.

420

421 If ``description`` is passed, the code is compiled against a synthetic

422 filename which is registered with :mod:`linecache`, allowing traceback

423 frames to reference the actual source code being referenced.

424

425 entries are placed in the cache without an expiration time and a

426 weakref.finalize() is applied to the function to remove the linecache

427 entry if and when the function is garbage collected.

428

429 """

430 filename: Optional[str] = None

431 entry: Optional[_LinecacheEntry] = None

432

433 if description is not None:

434 assert isinstance(code, str), (

435 "a description is only meaningful for source that has not "

436 "already been compiled"

437 )

438 filename = f"<sqlalchemy generated {description}>"

439

440 try:

441 linecache_cache = _linecache_cache_getter()

442 except AttributeError:

443 pass

444 else:

445 entry = (len(code), None, code.splitlines(True), filename)

446 linecache_cache[filename] = entry

447 code = compile(code, filename, "exec")

448

449 exec(code, env)

450 fn = env[fn_name]

451

452 if filename is not None and entry is not None:

453 # apply a finalizer that addresses on-the-fly functions, ORM mapped

454 # classes, etc. which might be garbage collected

455 finalizer = weakref.finalize(

456 fn, _remove_linecache_entry, filename, entry

457 )

458 # atexit is a property on the C implementation; typeshed

459 # renders finalize with an empty __slots__

460 finalizer.atexit = False # type: ignore[misc]

461

462 return fn # type: ignore[no-any-return]

463

464

465_PF = TypeVar("_PF")

466_TE = TypeVar("_TE")

467

468

469class PluginLoader:

470 def __init__(

471 self, group: str, auto_fn: Optional[Callable[..., Any]] = None

472 ):

473 self.group = group

474 self.impls: Dict[str, Any] = {}

475 self.auto_fn = auto_fn

476

477 def clear(self):

478 self.impls.clear()

479

480 def load(self, name: str) -> Any:

481 if name in self.impls:

482 return self.impls[name]()

483

484 if self.auto_fn:

485 loader = self.auto_fn(name)

486 if loader:

487 self.impls[name] = loader

488 return loader()

489

490 for impl in compat.importlib_metadata_get(self.group):

491 if impl.name == name:

492 self.impls[name] = impl.load

493 return impl.load()

494

495 raise exc.NoSuchModuleError(

496 "Can't load plugin: %s:%s" % (self.group, name)

497 )

498

499 def register(self, name: str, modulepath: str, objname: str) -> None:

500 def load():

501 mod = __import__(modulepath)

502 for token in modulepath.split(".")[1:]:

503 mod = getattr(mod, token)

504 return getattr(mod, objname)

505

506 self.impls[name] = load

507

508 def deregister(self, name: str) -> None:

509 del self.impls[name]

510

511

512def _inspect_func_args(fn):

513 try:

514 co_varkeywords = inspect.CO_VARKEYWORDS

515 except AttributeError:

516 # https://docs.python.org/3/library/inspect.html

517 # The flags are specific to CPython, and may not be defined in other

518 # Python implementations. Furthermore, the flags are an implementation

519 # detail, and can be removed or deprecated in future Python releases.

520 spec = compat.inspect_getfullargspec(fn)

521 return spec[0], bool(spec[2])

522 else:

523 # use fn.__code__ plus flags to reduce method call overhead

524 co = fn.__code__

525 nargs = co.co_argcount

526 return (

527 list(co.co_varnames[:nargs]),

528 bool(co.co_flags & co_varkeywords),

529 )

530

531

532@overload

533def get_cls_kwargs(

534 cls: type,

535 *,

536 _set: Optional[Set[str]] = None,

537 raiseerr: Literal[True] = ...,

538) -> Set[str]: ...

539

540

541@overload

542def get_cls_kwargs(

543 cls: type, *, _set: Optional[Set[str]] = None, raiseerr: bool = False

544) -> Optional[Set[str]]: ...

545

546

547def get_cls_kwargs(

548 cls: type, *, _set: Optional[Set[str]] = None, raiseerr: bool = False

549) -> Optional[Set[str]]:

550 r"""Return the full set of inherited kwargs for the given `cls`.

551

552 Probes a class's __init__ method, collecting all named arguments. If the

553 __init__ defines a \**kwargs catch-all, then the constructor is presumed

554 to pass along unrecognized keywords to its base classes, and the

555 collection process is repeated recursively on each of the bases.

556

557 Uses a subset of inspect.getfullargspec() to cut down on method overhead,

558 as this is used within the Core typing system to create copies of type

559 objects which is a performance-sensitive operation.

560

561 No anonymous tuple arguments please !

562

563 """

564 toplevel = _set is None

565 if toplevel:

566 _set = set()

567 assert _set is not None

568

569 ctr = cls.__dict__.get("__init__", False)

570

571 has_init = (

572 ctr

573 and isinstance(ctr, types.FunctionType)

574 and isinstance(ctr.__code__, types.CodeType)

575 )

576

577 if has_init:

578 names, has_kw = _inspect_func_args(ctr)

579 _set.update(names)

580

581 if not has_kw and not toplevel:

582 if raiseerr:

583 raise TypeError(

584 f"given cls {cls} doesn't have an __init__ method"

585 )

586 else:

587 return None

588 else:

589 has_kw = False

590

591 if not has_init or has_kw:

592 for c in cls.__bases__:

593 if get_cls_kwargs(c, _set=_set) is None:

594 break

595

596 _set.discard("self")

597 return _set

598

599

600def get_func_kwargs(func: Callable[..., Any]) -> List[str]:

601 """Return the set of legal kwargs for the given `func`.

602

603 Uses getargspec so is safe to call for methods, functions,

604 etc.

605

606 """

607

608 return compat.inspect_getfullargspec(func)[0]

609

610

611def get_callable_argspec(

612 fn: Callable[..., Any], no_self: bool = False, _is_init: bool = False

613) -> compat.FullArgSpec:

614 """Return the argument signature for any callable.

615

616 All pure-Python callables are accepted, including

617 functions, methods, classes, objects with __call__;

618 builtins and other edge cases like functools.partial() objects

619 raise a TypeError.

620

621 """

622 if inspect.isbuiltin(fn):

623 raise TypeError("Can't inspect builtin: %s" % fn)

624 elif inspect.isfunction(fn) or (

625 hasattr(fn, "__code__")

626 and not inspect.isclass(fn)

627 and not inspect.ismethod(fn)

628 ):

629 if _is_init and no_self:

630 spec = compat.inspect_getfullargspec(fn)

631 return compat.FullArgSpec(

632 spec.args[1:],

633 spec.varargs,

634 spec.varkw,

635 spec.defaults,

636 spec.kwonlyargs,

637 spec.kwonlydefaults,

638 spec.annotations,

639 )

640 else:

641 return compat.inspect_getfullargspec(fn)

642 elif inspect.ismethod(fn):

643 if no_self and (_is_init or fn.__self__):

644 spec = compat.inspect_getfullargspec(fn.__func__)

645 return compat.FullArgSpec(

646 spec.args[1:],

647 spec.varargs,

648 spec.varkw,

649 spec.defaults,

650 spec.kwonlyargs,

651 spec.kwonlydefaults,

652 spec.annotations,

653 )

654 else:

655 return compat.inspect_getfullargspec(fn.__func__)

656 elif inspect.isclass(fn):

657 return get_callable_argspec(

658 fn.__init__, no_self=no_self, _is_init=True

659 )

660 elif hasattr(fn, "__func__"):

661 return compat.inspect_getfullargspec(fn.__func__)

662 elif hasattr(fn, "__call__"):

663 if inspect.ismethod(fn.__call__):

664 return get_callable_argspec(fn.__call__, no_self=no_self)

665 else:

666 raise TypeError("Can't inspect callable: %s" % fn)

667 else:

668 raise TypeError("Can't inspect callable: %s" % fn)

669

670

671def format_argspec_plus(

672 fn: Union[Callable[..., Any], compat.FullArgSpec], grouped: bool = True

673) -> Dict[str, Optional[str]]:

674 """Returns a dictionary of formatted, introspected function arguments.

675

676 A enhanced variant of inspect.formatargspec to support code generation.

677

678 fn

679 An inspectable callable or tuple of inspect getargspec() results.

680 grouped

681 Defaults to True; include (parens, around, argument) lists

682

683 Returns:

684

685 args

686 Full inspect.formatargspec for fn

687 self_arg

688 The name of the first positional argument, varargs[0], or None

689 if the function defines no positional arguments.

690 apply_pos

691 args, re-written in calling rather than receiving syntax. Arguments are

692 passed positionally.

693 apply_kw

694 Like apply_pos, except keyword-ish args are passed as keywords.

695 apply_pos_proxied

696 Like apply_pos but omits the self/cls argument

697

698 Example::

699

700 >>> format_argspec_plus(lambda self, a, b, c=3, **d: 123)

701 {'grouped_args': '(self, a, b, c=3, **d)',

702 'self_arg': 'self',

703 'apply_kw': '(self, a, b, c=c, **d)',

704 'apply_pos': '(self, a, b, c, **d)'}

705

706 """

707 if callable(fn):

708 spec = compat.inspect_getfullargspec(fn)

709 else:

710 spec = fn

711

712 args = compat.inspect_formatargspec(*spec)

713

714 apply_pos = compat.inspect_formatargspec(

715 spec[0], spec[1], spec[2], None, spec[4]

716 )

717

718 if spec[0]:

719 self_arg = spec[0][0]

720

721 apply_pos_proxied = compat.inspect_formatargspec(

722 spec[0][1:], spec[1], spec[2], None, spec[4]

723 )

724

725 elif spec[1]:

726 # I'm not sure what this is

727 self_arg = "%s[0]" % spec[1]

728

729 apply_pos_proxied = apply_pos

730 else:

731 self_arg = None

732 apply_pos_proxied = apply_pos

733

734 num_defaults = 0

735 if spec[3]:

736 num_defaults += len(cast(Tuple[Any], spec[3]))

737 if spec[4]:

738 num_defaults += len(spec[4])

739

740 name_args = spec[0] + spec[4]

741

742 defaulted_vals: Union[List[str], Tuple[()]]

743

744 if num_defaults:

745 defaulted_vals = name_args[0 - num_defaults :]

746 else:

747 defaulted_vals = ()

748

749 apply_kw = compat.inspect_formatargspec(

750 name_args,

751 spec[1],

752 spec[2],

753 defaulted_vals,

754 formatvalue=lambda x: "=" + str(x),

755 )

756

757 if spec[0]:

758 apply_kw_proxied = compat.inspect_formatargspec(

759 name_args[1:],

760 spec[1],

761 spec[2],

762 defaulted_vals,

763 formatvalue=lambda x: "=" + str(x),

764 )

765 else:

766 apply_kw_proxied = apply_kw

767

768 if grouped:

769 return dict(

770 grouped_args=args,

771 self_arg=self_arg,

772 apply_pos=apply_pos,

773 apply_kw=apply_kw,

774 apply_pos_proxied=apply_pos_proxied,

775 apply_kw_proxied=apply_kw_proxied,

776 )

777 else:

778 return dict(

779 grouped_args=args,

780 self_arg=self_arg,

781 apply_pos=apply_pos[1:-1],

782 apply_kw=apply_kw[1:-1],

783 apply_pos_proxied=apply_pos_proxied[1:-1],

784 apply_kw_proxied=apply_kw_proxied[1:-1],

785 )

786

787

788def format_argspec_init(method, grouped=True):

789 """format_argspec_plus with considerations for typical __init__ methods

790

791 Wraps format_argspec_plus with error handling strategies for typical

792 __init__ cases:

793

794 .. sourcecode:: text

795

796 object.__init__ -> (self)

797 other unreflectable (usually C) -> (self, *args, **kwargs)

798

799 """

800 if method is object.__init__:

801 grouped_args = "(self)"

802 args = "(self)" if grouped else "self"

803 proxied = "()" if grouped else ""

804 else:

805 try:

806 return format_argspec_plus(method, grouped=grouped)

807 except TypeError:

808 grouped_args = "(self, *args, **kwargs)"

809 args = grouped_args if grouped else "self, *args, **kwargs"

810 proxied = "(*args, **kwargs)" if grouped else "*args, **kwargs"

811 return dict(

812 self_arg="self",

813 grouped_args=grouped_args,

814 apply_pos=args,

815 apply_kw=args,

816 apply_pos_proxied=proxied,

817 apply_kw_proxied=proxied,

818 )

819

820

821def create_proxy_methods(

822 target_cls: Type[Any],

823 target_cls_sphinx_name: str,

824 proxy_cls_sphinx_name: str,

825 classmethods: Sequence[str] = (),

826 methods: Sequence[str] = (),

827 attributes: Sequence[str] = (),

828 use_intermediate_variable: Sequence[str] = (),

829) -> Callable[[_T], _T]:

830 """A class decorator indicating attributes should refer to a proxy

831 class.

832

833 This decorator is now a "marker" that does nothing at runtime. Instead,

834 it is consumed by the tools/generate_proxy_methods.py script to

835 statically generate proxy methods and attributes that are fully

836 recognized by typing tools such as mypy.

837

838 """

839

840 def decorate(cls):

841 return cls

842

843 return decorate

844

845

846def getargspec_init(method):

847 """inspect.getargspec with considerations for typical __init__ methods

848

849 Wraps inspect.getargspec with error handling for typical __init__ cases:

850

851 .. sourcecode:: text

852

853 object.__init__ -> (self)

854 other unreflectable (usually C) -> (self, *args, **kwargs)

855

856 """

857 try:

858 return compat.inspect_getfullargspec(method)

859 except TypeError:

860 if method is object.__init__:

861 return (["self"], None, None, None)

862 else:

863 return (["self"], "args", "kwargs", None)

864

865

866def unbound_method_to_callable(func_or_cls):

867 """Adjust the incoming callable such that a 'self' argument is not

868 required.

869

870 """

871

872 if isinstance(func_or_cls, types.MethodType) and not func_or_cls.__self__:

873 return func_or_cls.__func__

874 else:

875 return func_or_cls

876

877

878class GenericRepr:

879 """Encapsulates the logic for creating a generic __repr__() string.

880

881 This class allows for the repr structure to be created, then modified

882 (e.g., changing the class name), before being rendered as a string.

883

884 .. versionadded:: 2.1

885 """

886

887 __slots__ = (

888 "_obj",

889 "_additional_kw",

890 "_to_inspect",

891 "_omit_kwarg",

892 "_class_name",

893 )

894

895 _obj: Any

896 _additional_kw: Sequence[Tuple[str, Any]]

897 _to_inspect: List[object]

898 _omit_kwarg: Sequence[str]

899 _class_name: Optional[str]

900

901 def __init__(

902 self,

903 obj: Any,

904 additional_kw: Sequence[Tuple[str, Any]] = (),

905 to_inspect: Optional[Union[object, List[object]]] = None,

906 omit_kwarg: Sequence[str] = (),

907 ):

908 """Create a GenericRepr object.

909

910 :param obj: The object being repr'd

911 :param additional_kw: Additional keyword arguments to check for in

912 the repr, as a sequence of 2-tuples of (name, default_value)

913 :param to_inspect: One or more objects whose __init__ signature

914 should be inspected. If not provided, defaults to [obj].

915 :param omit_kwarg: Sequence of keyword argument names to omit from

916 the repr output

917 """

918 self._obj = obj

919 self._additional_kw = additional_kw

920 self._to_inspect = (

921 [obj] if to_inspect is None else _collections.to_list(to_inspect)

922 )

923 self._omit_kwarg = omit_kwarg

924 self._class_name = None

925

926 def set_class_name(self, class_name: str) -> GenericRepr:

927 """Set the class name to be used in the repr.

928

929 By default, the class name is taken from obj.__class__.__name__.

930 This method allows it to be overridden.

931

932 :param class_name: The class name to use

933 :return: self, for method chaining

934 """

935 self._class_name = class_name

936 return self

937

938 def __str__(self) -> str:

939 """Produce the __repr__() string based on the configured parameters."""

940 obj = self._obj

941 to_inspect = self._to_inspect

942 additional_kw = self._additional_kw

943 omit_kwarg = self._omit_kwarg

944

945 missing = object()

946

947 pos_args = []

948 kw_args: _collections.OrderedDict[str, Any] = (

949 _collections.OrderedDict()

950 )

951 vargs = None

952 for i, insp in enumerate(to_inspect):

953 try:

954 spec = compat.inspect_getfullargspec(insp.__init__) # type: ignore[misc] # noqa: E501

955 except TypeError:

956 continue

957 else:

958 default_len = len(spec.defaults) if spec.defaults else 0

959 if i == 0:

960 if spec.varargs:

961 vargs = spec.varargs

962 if default_len:

963 pos_args.extend(spec.args[1:-default_len])

964 else:

965 pos_args.extend(spec.args[1:])

966 else:

967 kw_args.update(

968 [(arg, missing) for arg in spec.args[1:-default_len]]

969 )

970

971 if default_len:

972 assert spec.defaults

973 kw_args.update(

974 [

975 (arg, default)

976 for arg, default in zip(

977 spec.args[-default_len:], spec.defaults

978 )

979 ]

980 )

981 output: List[str] = []

982

983 output.extend(repr(getattr(obj, arg, None)) for arg in pos_args)

984

985 if vargs is not None and hasattr(obj, vargs):

986 output.extend([repr(val) for val in getattr(obj, vargs)])

987

988 for arg, defval in kw_args.items():

989 if arg in omit_kwarg:

990 continue

991 try:

992 val = getattr(obj, arg, missing)

993 if val is not missing and val != defval:

994 output.append("%s=%r" % (arg, val))

995 except Exception:

996 pass

997

998 if additional_kw:

999 for arg, defval in additional_kw:

1000 try:

1001 val = getattr(obj, arg, missing)

1002 if val is not missing and val != defval:

1003 output.append("%s=%r" % (arg, val))

1004 except Exception:

1005 pass

1006

1007 class_name = (

1008 self._class_name

1009 if self._class_name is not None

1010 else obj.__class__.__name__

1011 )

1012 return "%s(%s)" % (class_name, ", ".join(output))

1013

1014

1015def generic_repr(

1016 obj: Any,

1017 additional_kw: Sequence[Tuple[str, Any]] = (),

1018 to_inspect: Optional[Union[object, List[object]]] = None,

1019 omit_kwarg: Sequence[str] = (),

1020) -> str:

1021 """Produce a __repr__() based on direct association of the __init__()

1022 specification vs. same-named attributes present.

1023

1024 """

1025 return str(

1026 GenericRepr(

1027 obj,

1028 additional_kw=additional_kw,

1029 to_inspect=to_inspect,

1030 omit_kwarg=omit_kwarg,

1031 )

1032 )

1033

1034

1035def class_hierarchy(cls):

1036 """Return an unordered sequence of all classes related to cls.

1037

1038 Traverses diamond hierarchies.

1039

1040 Fibs slightly: subclasses of builtin types are not returned. Thus

1041 class_hierarchy(class A(object)) returns (A, object), not A plus every

1042 class systemwide that derives from object.

1043

1044 """

1045

1046 hier = {cls}

1047 process = list(cls.__mro__)

1048 while process:

1049 c = process.pop()

1050 bases = (_ for _ in c.__bases__ if _ not in hier)

1051

1052 for b in bases:

1053 process.append(b)

1054 hier.add(b)

1055

1056 if c.__module__ == "builtins" or not hasattr(c, "__subclasses__"):

1057 continue

1058

1059 for s in [

1060 _

1061 for _ in (

1062 c.__subclasses__()

1063 if not issubclass(c, type)

1064 else c.__subclasses__(c)

1065 )

1066 if _ not in hier

1067 ]:

1068 process.append(s)

1069 hier.add(s)

1070 return list(hier)

1071

1072

1073def iterate_attributes(cls):

1074 """iterate all the keys and attributes associated

1075 with a class, without using getattr().

1076

1077 Does not use getattr() so that class-sensitive

1078 descriptors (i.e. property.__get__()) are not called.

1079

1080 """

1081 keys = dir(cls)

1082 for key in keys:

1083 for c in cls.__mro__:

1084 if key in c.__dict__:

1085 yield (key, c.__dict__[key])

1086 break

1087

1088

1089def monkeypatch_proxied_specials(

1090 into_cls,

1091 from_cls,

1092 skip=None,

1093 only=None,

1094 name="self.proxy",

1095 from_instance=None,

1096):

1097 """Automates delegation of __specials__ for a proxying type."""

1098

1099 if only:

1100 dunders = only

1101 else:

1102 if skip is None:

1103 skip = (

1104 "__slots__",

1105 "__del__",

1106 "__getattribute__",

1107 "__metaclass__",

1108 "__getstate__",

1109 "__setstate__",

1110 )

1111 dunders = [

1112 m

1113 for m in dir(from_cls)

1114 if (

1115 m.startswith("__")

1116 and m.endswith("__")

1117 and not hasattr(into_cls, m)

1118 and m not in skip

1119 )

1120 ]

1121

1122 for method in dunders:

1123 try:

1124 maybe_fn = getattr(from_cls, method)

1125 if not hasattr(maybe_fn, "__call__"):

1126 continue

1127 maybe_fn = getattr(maybe_fn, "__func__", maybe_fn)

1128 fn = cast(types.FunctionType, maybe_fn)

1129

1130 except AttributeError:

1131 continue

1132 try:

1133 spec = compat.inspect_getfullargspec(fn)

1134 fn_args = compat.inspect_formatargspec(spec[0])

1135 d_args = compat.inspect_formatargspec(spec[0][1:])

1136 except TypeError:

1137 fn_args = "(self, *args, **kw)"

1138 d_args = "(*args, **kw)"

1139

1140 py = (

1141 "def %(method)s%(fn_args)s: "

1142 "return %(name)s.%(method)s%(d_args)s" % locals()

1143 )

1144

1145 env: Dict[str, types.FunctionType] = (

1146 from_instance is not None and {name: from_instance} or {}

1147 )

1148 # the generated source is derived entirely from from_cls and the method

1149 # name, and into_cls is a throwaway per-descriptor class whose

1150 # __qualname__ is the same for every one of them, so this is keyed on

1151 # from_cls rather than into_cls

1152 proxied = exec_code_in_env(

1153 py,

1154 env,

1155 method,

1156 f"{method} proxying to "

1157 f"{from_cls.__module__}.{from_cls.__qualname__}",

1158 )

1159 try:

1160 proxied.__defaults__ = fn.__defaults__

1161 except AttributeError:

1162 pass

1163 setattr(into_cls, method, proxied)

1164

1165

1166def methods_equivalent(meth1, meth2):

1167 """Return True if the two methods are the same implementation."""

1168

1169 return getattr(meth1, "__func__", meth1) is getattr(

1170 meth2, "__func__", meth2

1171 )

1172

1173

1174def as_interface(obj, cls=None, methods=None, required=None):

1175 """Ensure basic interface compliance for an instance or dict of callables.

1176

1177 Checks that ``obj`` implements public methods of ``cls`` or has members

1178 listed in ``methods``. If ``required`` is not supplied, implementing at

1179 least one interface method is sufficient. Methods present on ``obj`` that

1180 are not in the interface are ignored.

1181

1182 If ``obj`` is a dict and ``dict`` does not meet the interface

1183 requirements, the keys of the dictionary are inspected. Keys present in

1184 ``obj`` that are not in the interface will raise TypeErrors.

1185

1186 Raises TypeError if ``obj`` does not meet the interface criteria.

1187

1188 In all passing cases, an object with callable members is returned. In the

1189 simple case, ``obj`` is returned as-is; if dict processing kicks in then

1190 an anonymous class is returned.

1191

1192 obj

1193 A type, instance, or dictionary of callables.

1194 cls

1195 Optional, a type. All public methods of cls are considered the

1196 interface. An ``obj`` instance of cls will always pass, ignoring

1197 ``required``..

1198 methods

1199 Optional, a sequence of method names to consider as the interface.

1200 required

1201 Optional, a sequence of mandatory implementations. If omitted, an

1202 ``obj`` that provides at least one interface method is considered

1203 sufficient. As a convenience, required may be a type, in which case

1204 all public methods of the type are required.

1205

1206 """

1207 if not cls and not methods:

1208 raise TypeError("a class or collection of method names are required")

1209

1210 if isinstance(cls, type) and isinstance(obj, cls):

1211 return obj

1212

1213 interface = set(methods or [m for m in dir(cls) if not m.startswith("_")])

1214 implemented = set(dir(obj))

1215

1216 complies = operator.ge

1217 if isinstance(required, type):

1218 required = interface

1219 elif not required:

1220 required = set()

1221 complies = operator.gt

1222 else:

1223 required = set(required)

1224

1225 if complies(implemented.intersection(interface), required):

1226 return obj

1227

1228 # No dict duck typing here.

1229 if not isinstance(obj, dict):

1230 qualifier = complies is operator.gt and "any of" or "all of"

1231 raise TypeError(

1232 "%r does not implement %s: %s"

1233 % (obj, qualifier, ", ".join(interface))

1234 )

1235

1236 class AnonymousInterface:

1237 """A callable-holding shell."""

1238

1239 if cls:

1240 AnonymousInterface.__name__ = "Anonymous" + cls.__name__

1241 found = set()

1242

1243 for method, impl in dictlike_iteritems(obj):

1244 if method not in interface:

1245 raise TypeError("%r: unknown in this interface" % method)

1246 if not callable(impl):

1247 raise TypeError("%r=%r is not callable" % (method, impl))

1248 setattr(AnonymousInterface, method, staticmethod(impl))

1249 found.add(method)

1250

1251 if complies(found, required):

1252 return AnonymousInterface

1253

1254 raise TypeError(

1255 "dictionary does not contain required keys %s"

1256 % ", ".join(required - found)

1257 )

1258

1259

1260_GFD = TypeVar("_GFD", bound="generic_fn_descriptor[Any]")

1261

1262

1263class generic_fn_descriptor(Generic[_T_co]):

1264 """Descriptor which proxies a function when the attribute is not

1265 present in dict

1266

1267 This superclass is organized in a particular way with "memoized" and

1268 "non-memoized" implementation classes that are hidden from type checkers,

1269 as Mypy seems to not be able to handle seeing multiple kinds of descriptor

1270 classes used for the same attribute.

1271

1272 """

1273

1274 fget: Callable[..., _T_co]

1275 __doc__: Optional[str]

1276 __name__: str

1277

1278 def __init__(self, fget: Callable[..., _T_co], doc: Optional[str] = None):

1279 self.fget = fget

1280 self.__doc__ = doc or fget.__doc__

1281 self.__name__ = fget.__name__

1282

1283 @overload

1284 def __get__(self: _GFD, obj: None, cls: Any) -> _GFD: ...

1285

1286 @overload

1287 def __get__(self, obj: object, cls: Any) -> _T_co: ...

1288

1289 def __get__(self: _GFD, obj: Any, cls: Any) -> Union[_GFD, _T_co]:

1290 raise NotImplementedError()

1291

1292 if TYPE_CHECKING:

1293

1294 def __set__(self, instance: Any, value: Any) -> None: ...

1295

1296 def __delete__(self, instance: Any) -> None: ...

1297

1298 def _reset(self, obj: Any) -> None:

1299 raise NotImplementedError()

1300

1301 @classmethod

1302 def reset(cls, obj: Any, name: str) -> None:

1303 raise NotImplementedError()

1304

1305

1306class _non_memoized_property(generic_fn_descriptor[_T_co]):

1307 """a plain descriptor that proxies a function.

1308

1309 primary rationale is to provide a plain attribute that's

1310 compatible with memoized_property which is also recognized as equivalent

1311 by mypy.

1312

1313 """

1314

1315 if not TYPE_CHECKING:

1316

1317 def __get__(self, obj, cls):

1318 if obj is None:

1319 return self

1320 return self.fget(obj)

1321

1322

1323class _memoized_property(generic_fn_descriptor[_T_co]):

1324 """A read-only @property that is only evaluated once."""

1325

1326 if not TYPE_CHECKING:

1327

1328 def __get__(self, obj, cls):

1329 if obj is None:

1330 return self

1331 obj.__dict__[self.__name__] = result = self.fget(obj)

1332 return result

1333

1334 def _reset(self, obj):

1335 _memoized_property.reset(obj, self.__name__)

1336

1337 @classmethod

1338 def reset(cls, obj, name):

1339 obj.__dict__.pop(name, None)

1340

1341

1342# despite many attempts to get Mypy to recognize an overridden descriptor

1343# where one is memoized and the other isn't, there seems to be no reliable

1344# way other than completely deceiving the type checker into thinking there

1345# is just one single descriptor type everywhere. Otherwise, if a superclass

1346# has non-memoized and subclass has memoized, that requires

1347# "class memoized(non_memoized)". but then if a superclass has memoized and

1348# superclass has non-memoized, the class hierarchy of the descriptors

1349# would need to be reversed; "class non_memoized(memoized)". so there's no

1350# way to achieve this.

1351# additional issues, RO properties:

1352# https://github.com/python/mypy/issues/12440

1353if TYPE_CHECKING:

1354 # allow memoized and non-memoized to be freely mixed by having them

1355 # be the same class

1356 memoized_property = generic_fn_descriptor

1357 non_memoized_property = generic_fn_descriptor

1358

1359 # for read only situations, mypy only sees @property as read only.

1360 # read only is needed when a subtype specializes the return type

1361 # of a property, meaning assignment needs to be disallowed

1362 ro_memoized_property = property

1363 ro_non_memoized_property = property

1364

1365else:

1366 memoized_property = ro_memoized_property = _memoized_property

1367 non_memoized_property = ro_non_memoized_property = _non_memoized_property

1368

1369

1370def memoized_instancemethod(fn: _F) -> _F:

1371 """Decorate a method memoize its return value.

1372

1373 Best applied to no-arg methods: memoization is not sensitive to

1374 argument values, and will always return the same value even when

1375 called with different arguments.

1376

1377 """

1378

1379 def oneshot(self, *args, **kw):

1380 result = fn(self, *args, **kw)

1381

1382 def memo(*a, **kw):

1383 return result

1384

1385 memo.__name__ = fn.__name__

1386 memo.__doc__ = fn.__doc__

1387 self.__dict__[fn.__name__] = memo

1388 return result

1389

1390 return update_wrapper(oneshot, fn) # type: ignore[return-value]

1391

1392

1393class HasMemoized:

1394 """A mixin class that maintains the names of memoized elements in a

1395 collection for easy cache clearing, generative, etc.

1396

1397 """

1398

1399 if not TYPE_CHECKING:

1400 # support classes that want to have __slots__ with an explicit

1401 # slot for __dict__. not sure if that requires base __slots__ here.

1402 __slots__ = ()

1403

1404 _memoized_keys: FrozenSet[str] = frozenset()

1405

1406 def _reset_memoizations(self) -> None:

1407 for elem in self._memoized_keys:

1408 self.__dict__.pop(elem, None)

1409

1410 def _assert_no_memoizations(self) -> None:

1411 for elem in self._memoized_keys:

1412 assert elem not in self.__dict__

1413

1414 def _set_memoized_attribute(self, key: str, value: Any) -> None:

1415 self.__dict__[key] = value

1416 self._memoized_keys |= {key}

1417

1418 class memoized_attribute(memoized_property[_T]):

1419 """A read-only @property that is only evaluated once.

1420

1421 :meta private:

1422

1423 """

1424

1425 fget: Callable[..., _T]

1426 __doc__: Optional[str]

1427 __name__: str

1428

1429 def __init__(self, fget: Callable[..., _T], doc: Optional[str] = None):

1430 self.fget = fget

1431 self.__doc__ = doc or fget.__doc__

1432 self.__name__ = fget.__name__

1433

1434 @overload

1435 def __get__(self: _MA, obj: None, cls: Any) -> _MA: ...

1436

1437 @overload

1438 def __get__(self, obj: Any, cls: Any) -> _T: ...

1439

1440 def __get__(self, obj, cls):

1441 if obj is None:

1442 return self

1443 obj.__dict__[self.__name__] = result = self.fget(obj)

1444 obj._memoized_keys |= {self.__name__}

1445 return result

1446

1447 @classmethod

1448 def memoized_instancemethod(cls, fn: _F) -> _F:

1449 """Decorate a method memoize its return value.

1450

1451 :meta private:

1452

1453 """

1454

1455 def oneshot(self: Any, *args: Any, **kw: Any) -> Any:

1456 result = fn(self, *args, **kw)

1457

1458 def memo(*a, **kw):

1459 return result

1460

1461 memo.__name__ = fn.__name__

1462 memo.__doc__ = fn.__doc__

1463 self.__dict__[fn.__name__] = memo

1464 self._memoized_keys |= {fn.__name__}

1465 return result

1466

1467 return update_wrapper(oneshot, fn) # type: ignore[return-value]

1468

1469

1470if TYPE_CHECKING:

1471 HasMemoized_ro_memoized_attribute = property

1472else:

1473 HasMemoized_ro_memoized_attribute = HasMemoized.memoized_attribute

1474

1475

1476class MemoizedSlots:

1477 """Apply memoized items to an object using a __getattr__ scheme.

1478

1479 This allows the functionality of memoized_property and

1480 memoized_instancemethod to be available to a class using __slots__.

1481

1482 The memoized get is not threadsafe under freethreading and the

1483 creator method may in extremely rare cases be called more than once.

1484

1485 """

1486

1487 __slots__ = ()

1488

1489 def _fallback_getattr(self, key):

1490 raise AttributeError(key)

1491

1492 def __getattr__(self, key: str) -> Any:

1493 if key.startswith("_memoized_attr_") or key.startswith(

1494 "_memoized_method_"

1495 ):

1496 raise AttributeError(key)

1497 # to avoid recursion errors when interacting with other __getattr__

1498 # schemes that refer to this one, when testing for memoized method

1499 # look at __class__ only rather than going into __getattr__ again.

1500 elif hasattr(self.__class__, f"_memoized_attr_{key}"):

1501 value = getattr(self, f"_memoized_attr_{key}")()

1502 setattr(self, key, value)

1503 return value

1504 elif hasattr(self.__class__, f"_memoized_method_{key}"):

1505 meth = getattr(self, f"_memoized_method_{key}")

1506

1507 def oneshot(*args, **kw):

1508 result = meth(*args, **kw)

1509

1510 def memo(*a, **kw):

1511 return result

1512

1513 memo.__name__ = meth.__name__

1514 memo.__doc__ = meth.__doc__

1515 setattr(self, key, memo)

1516 return result

1517

1518 oneshot.__doc__ = meth.__doc__

1519 return oneshot

1520 else:

1521 return self._fallback_getattr(key)

1522

1523

1524# from paste.deploy.converters

1525def asbool(obj: Any) -> bool:

1526 if isinstance(obj, str):

1527 obj = obj.strip().lower()

1528 if obj in ["true", "yes", "on", "y", "t", "1"]:

1529 return True

1530 elif obj in ["false", "no", "off", "n", "f", "0"]:

1531 return False

1532 else:

1533 raise ValueError("String is not true/false: %r" % obj)

1534 return bool(obj)

1535

1536

1537def bool_or_str(*text: str) -> Callable[[str], Union[str, bool]]:

1538 """Return a callable that will evaluate a string as

1539 boolean, or one of a set of "alternate" string values.

1540

1541 """

1542

1543 def bool_or_value(obj: str) -> Union[str, bool]:

1544 if obj in text:

1545 return obj

1546 else:

1547 return asbool(obj)

1548

1549 return bool_or_value

1550

1551

1552def asint(value: Any) -> Optional[int]:

1553 """Coerce to integer."""

1554

1555 if value is None:

1556 return value

1557 return int(value)

1558

1559

1560def coerce_kw_type(

1561 kw: Dict[str, Any],

1562 key: str,

1563 type_: Type[Any],

1564 flexi_bool: bool = True,

1565 dest: Optional[Dict[str, Any]] = None,

1566) -> None:

1567 r"""If 'key' is present in dict 'kw', coerce its value to type 'type\_' if

1568 necessary. If 'flexi_bool' is True, the string '0' is considered false

1569 when coercing to boolean.

1570 """

1571

1572 if dest is None:

1573 dest = kw

1574

1575 if (

1576 key in kw

1577 and (not isinstance(type_, type) or not isinstance(kw[key], type_))

1578 and kw[key] is not None

1579 ):

1580 if type_ is bool and flexi_bool:

1581 dest[key] = asbool(kw[key])

1582 else:

1583 dest[key] = type_(kw[key])

1584

1585

1586def constructor_key(obj: Any, cls: Type[Any]) -> Tuple[Any, ...]:

1587 """Produce a tuple structure that is cacheable using the __dict__ of

1588 obj to retrieve values

1589

1590 """

1591 names = get_cls_kwargs(cls)

1592 return (cls,) + tuple(

1593 (k, obj.__dict__[k]) for k in names if k in obj.__dict__

1594 )

1595

1596

1597def constructor_copy(obj: _T, cls: Type[_T], *args: Any, **kw: Any) -> _T:

1598 """Instantiate cls using the __dict__ of obj as constructor arguments.

1599

1600 Uses inspect to match the named arguments of ``cls``.

1601

1602 """

1603

1604 names = get_cls_kwargs(cls)

1605 kw.update(

1606 (k, obj.__dict__[k]) for k in names.difference(kw) if k in obj.__dict__

1607 )

1608 return cls(*args, **kw)

1609

1610

1611def counter() -> Callable[[], int]:

1612 """Return a threadsafe counter function."""

1613

1614 lock = threading.Lock()

1615 counter = itertools.count(1)

1616

1617 # avoid the 2to3 "next" transformation...

1618 def _next():

1619 with lock:

1620 return next(counter)

1621

1622 return _next

1623

1624

1625def duck_type_collection(

1626 specimen: Any, default: Optional[Type[Any]] = None

1627) -> Optional[Type[Any]]:

1628 """Given an instance or class, guess if it is or is acting as one of

1629 the basic collection types: list, set and dict. If the __emulates__

1630 property is present, return that preferentially.

1631 """

1632

1633 if hasattr(specimen, "__emulates__"):

1634 # canonicalize set vs sets.Set to a standard: the builtin set

1635 if specimen.__emulates__ is not None and issubclass(

1636 specimen.__emulates__, set

1637 ):

1638 return set

1639 else:

1640 return specimen.__emulates__ # type: ignore[no-any-return]

1641

1642 isa = issubclass if isinstance(specimen, type) else isinstance

1643 if isa(specimen, list):

1644 return list

1645 elif isa(specimen, set):

1646 return set

1647 elif isa(specimen, dict):

1648 return dict

1649

1650 if hasattr(specimen, "append"):

1651 return list

1652 elif hasattr(specimen, "add"):

1653 return set

1654 elif hasattr(specimen, "set"):

1655 return dict

1656 else:

1657 return default

1658

1659

1660def assert_arg_type(

1661 arg: Any, argtype: Union[Tuple[Type[Any], ...], Type[Any]], name: str

1662) -> Any:

1663 if isinstance(arg, argtype):

1664 return arg

1665 else:

1666 if isinstance(argtype, tuple):

1667 raise exc.ArgumentError(

1668 "Argument '%s' is expected to be one of type %s, got '%s'"

1669 % (name, " or ".join("'%s'" % a for a in argtype), type(arg))

1670 )

1671 else:

1672 raise exc.ArgumentError(

1673 "Argument '%s' is expected to be of type '%s', got '%s'"

1674 % (name, argtype, type(arg))

1675 )

1676

1677

1678def dictlike_iteritems(dictlike):

1679 """Return a (key, value) iterator for almost any dict-like object."""

1680

1681 if hasattr(dictlike, "items"):

1682 return list(dictlike.items())

1683

1684 getter = getattr(dictlike, "__getitem__", getattr(dictlike, "get", None))

1685 if getter is None:

1686 raise TypeError("Object '%r' is not dict-like" % dictlike)

1687

1688 if hasattr(dictlike, "iterkeys"):

1689

1690 def iterator():

1691 for key in dictlike.iterkeys():

1692 assert getter is not None

1693 yield key, getter(key)

1694

1695 return iterator()

1696 elif hasattr(dictlike, "keys"):

1697 return iter((key, getter(key)) for key in dictlike.keys())

1698 else:

1699 raise TypeError("Object '%r' is not dict-like" % dictlike)

1700

1701

1702class classproperty(property):

1703 """A decorator that behaves like @property except that operates

1704 on classes rather than instances.

1705

1706 The decorator is currently special when using the declarative

1707 module, but note that the

1708 :class:`~.sqlalchemy.ext.declarative.declared_attr`

1709 decorator should be used for this purpose with declarative.

1710

1711 """

1712

1713 fget: Callable[[Any], Any]

1714

1715 def __init__(self, fget: Callable[[Any], Any], *arg: Any, **kw: Any):

1716 super().__init__(fget, *arg, **kw)

1717 self.__doc__ = fget.__doc__

1718

1719 def __get__(self, obj: Any, cls: Optional[type] = None) -> Any:

1720 return self.fget(cls)

1721

1722

1723class hybridproperty(Generic[_T]):

1724 def __init__(self, func: Callable[..., _T]):

1725 self.func = func

1726 self.clslevel = func

1727

1728 def __get__(self, instance: Any, owner: Any) -> _T:

1729 if instance is None:

1730 clsval = self.clslevel(owner)

1731 return clsval

1732 else:

1733 return self.func(instance)

1734

1735 def classlevel(self, func: Callable[..., Any]) -> hybridproperty[_T]:

1736 self.clslevel = func

1737 return self

1738

1739

1740class rw_hybridproperty(Generic[_T]):

1741 def __init__(self, func: Callable[..., _T]):

1742 self.func = func

1743 self.clslevel = func

1744 self.setfn: Optional[Callable[..., Any]] = None

1745

1746 def __get__(self, instance: Any, owner: Any) -> _T:

1747 if instance is None:

1748 clsval = self.clslevel(owner)

1749 return clsval

1750 else:

1751 return self.func(instance)

1752

1753 def __set__(self, instance: Any, value: Any) -> None:

1754 assert self.setfn is not None

1755 self.setfn(instance, value)

1756

1757 def setter(self, func: Callable[..., Any]) -> rw_hybridproperty[_T]:

1758 self.setfn = func

1759 return self

1760

1761 def classlevel(self, func: Callable[..., Any]) -> rw_hybridproperty[_T]:

1762 self.clslevel = func

1763 return self

1764

1765

1766class hybridmethod(Generic[_T]):

1767 """Decorate a function as cls- or instance- level."""

1768

1769 def __init__(self, func: Callable[..., _T]):

1770 self.func = self.__func__ = func

1771 self.clslevel = func

1772

1773 def __get__(self, instance: Any, owner: Any) -> Callable[..., _T]:

1774 if instance is None:

1775 return self.clslevel.__get__( # type: ignore[no-any-return]

1776 owner, owner.__class__

1777 )

1778 else:

1779 return self.func.__get__( # type: ignore[no-any-return]

1780 instance, owner

1781 )

1782

1783 def classlevel(self, func: Callable[..., Any]) -> hybridmethod[_T]:

1784 self.clslevel = func

1785 return self

1786

1787

1788class symbol(int):

1789 """A constant symbol.

1790

1791 >>> symbol("foo") is symbol("foo")

1792 True

1793 >>> symbol("foo")

1794 <symbol 'foo>

1795

1796 A slight refinement of the MAGICCOOKIE=object() pattern. The primary

1797 advantage of symbol() is its repr(). They are also singletons.

1798

1799 Repeated calls of symbol('name') will all return the same instance.

1800

1801 """

1802

1803 name: str

1804

1805 symbols: Dict[str, symbol] = {}

1806 _lock = threading.Lock()

1807

1808 def __new__(

1809 cls,

1810 name: str,

1811 doc: Optional[str] = None,

1812 canonical: Optional[int] = None,

1813 ) -> symbol:

1814 with cls._lock:

1815 sym = cls.symbols.get(name)

1816 if sym is None:

1817 assert isinstance(name, str)

1818 if canonical is None:

1819 canonical = hash(name)

1820 sym = int.__new__(symbol, canonical)

1821 sym.name = name

1822 if doc:

1823 sym.__doc__ = doc

1824

1825 # NOTE: we should ultimately get rid of this global thing,

1826 # however, currently it is to support pickling. The best

1827 # change would be when we are on py3.11 at a minimum, we

1828 # switch to stdlib enum.IntFlag.

1829 cls.symbols[name] = sym

1830 else:

1831 if canonical and canonical != sym:

1832 raise TypeError(

1833 f"Can't replace canonical symbol for {name!r} "

1834 f"with new int value {canonical}"

1835 )

1836 return sym

1837

1838 def __reduce__(self):

1839 return symbol, (self.name, "x", int(self))

1840

1841 def __str__(self):

1842 return repr(self)

1843

1844 def __repr__(self):

1845 return f"symbol({self.name!r})"

1846

1847

1848class _IntFlagMeta(type):

1849 def __init__(

1850 cls,

1851 classname: str,

1852 bases: Tuple[Type[Any], ...],

1853 dict_: Dict[str, Any],

1854 **kw: Any,

1855 ) -> None:

1856 items: List[symbol]

1857 cls._items = items = []

1858 for k, v in dict_.items():

1859 if re.match(r"^__.*__$", k):

1860 continue

1861 if isinstance(v, int):

1862 sym = symbol(k, canonical=v)

1863 elif not k.startswith("_"):

1864 raise TypeError("Expected integer values for IntFlag")

1865 else:

1866 continue

1867 setattr(cls, k, sym)

1868 items.append(sym)

1869

1870 cls.__members__ = _collections.immutabledict(

1871 {sym.name: sym for sym in items}

1872 )

1873

1874 def __iter__(self) -> Iterator[symbol]:

1875 raise NotImplementedError(

1876 "iter not implemented to ensure compatibility with "

1877 "Python 3.11 IntFlag. Please use __members__. See "

1878 "https://github.com/python/cpython/issues/99304"

1879 )

1880

1881

1882class _FastIntFlag(metaclass=_IntFlagMeta):

1883 """An 'IntFlag' copycat that isn't slow when performing bitwise

1884 operations.

1885

1886 the ``FastIntFlag`` class will return ``enum.IntFlag`` under TYPE_CHECKING

1887 and ``_FastIntFlag`` otherwise.

1888

1889 """

1890

1891

1892if TYPE_CHECKING:

1893 from enum import IntFlag

1894

1895 FastIntFlag = IntFlag

1896else:

1897 FastIntFlag = _FastIntFlag

1898

1899

1900_E = TypeVar("_E", bound=enum.Enum)

1901

1902

1903def parse_user_argument_for_enum(

1904 arg: Any,

1905 choices: Dict[_E, List[Any]],

1906 name: str,

1907 resolve_symbol_names: bool = False,

1908) -> Optional[_E]:

1909 """Given a user parameter, parse the parameter into a chosen value

1910 from a list of choice objects, typically Enum values.

1911

1912 The user argument can be a string name that matches the name of a

1913 symbol, or the symbol object itself, or any number of alternate choices

1914 such as True/False/ None etc.

1915

1916 :param arg: the user argument.

1917 :param choices: dictionary of enum values to lists of possible

1918 entries for each.

1919 :param name: name of the argument. Used in an :class:`.ArgumentError`

1920 that is raised if the parameter doesn't match any available argument.

1921

1922 """

1923 for enum_value, choice in choices.items():

1924 if arg is enum_value:

1925 return enum_value

1926 elif resolve_symbol_names and arg == enum_value.name:

1927 return enum_value

1928 elif arg in choice:

1929 return enum_value

1930

1931 if arg is None:

1932 return None

1933

1934 raise exc.ArgumentError(f"Invalid value for '{name}': {arg!r}")

1935

1936

1937_creation_order = 1

1938

1939

1940def set_creation_order(instance: Any) -> None:

1941 """Assign a '_creation_order' sequence to the given instance.

1942

1943 This allows multiple instances to be sorted in order of creation

1944 (typically within a single thread; the counter is not particularly

1945 threadsafe).

1946

1947 """

1948 global _creation_order

1949 instance._creation_order = _creation_order

1950 _creation_order += 1

1951

1952

1953def warn_exception(func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:

1954 """executes the given function, catches all exceptions and converts to

1955 a warning.

1956

1957 """

1958 try:

1959 return func(*args, **kwargs)

1960 except Exception:

1961 warn("%s('%s') ignored" % sys.exc_info()[0:2])

1962

1963

1964def ellipses_string(value, len_=25):

1965 try:

1966 if len(value) > len_:

1967 return "%s..." % value[0:len_]

1968 else:

1969 return value

1970 except TypeError:

1971 return value

1972

1973

1974class _hash_limit_string(str):

1975 """A string subclass that can only be hashed on a maximum amount

1976 of unique values.

1977

1978 This is used for warnings so that we can send out parameterized warnings

1979 without the __warningregistry__ of the module, or the non-overridable

1980 "once" registry within warnings.py, overloading memory,

1983 """

1985 _hash: int

1987 def __new__(

1988 cls, value: str, num: int, args: Sequence[Any]

1989 ) -> _hash_limit_string:

1990 interpolated = (value % args) + (

1991 " (this warning may be suppressed after %d occurrences)" % num

1992 )

1993 self = super().__new__(cls, interpolated)

1994 self._hash = hash("%s_%d" % (value, hash(interpolated) % num))

1995 return self

1996

1997 def __hash__(self) -> int:

1998 return self._hash

1999

2000 def __eq__(self, other: Any) -> bool:

2001 return hash(self) == hash(other)

2002

2003

2004def warn(msg: str, code: Optional[str] = None) -> None:

2005 """Issue a warning.

2006

2007 If msg is a string, :class:`.exc.SAWarning` is used as

2008 the category.

2009

2010 """

2011 if code:

2012 _warnings_warn(exc.SAWarning(msg, code=code))

2013 else:

2014 _warnings_warn(msg, exc.SAWarning)

2015

2016

2017def warn_limited(msg: str, args: Sequence[Any]) -> None:

2018 """Issue a warning with a parameterized string, limiting the number

2019 of registrations.

2020

2021 """

2022 if args:

2023 msg = _hash_limit_string(msg, 10, args)

2024 _warnings_warn(msg, exc.SAWarning)

2025

2026

2027_warning_tags: Dict[CodeType, Tuple[str, Type[Warning]]] = {}

2028

2029

2030def tag_method_for_warnings(

2031 message: str, category: Type[Warning]

2032) -> Callable[[_F], _F]:

2033 def go(fn):

2034 _warning_tags[fn.__code__] = (message, category)

2035 return fn

2036

2037 return go

2038

2039

2040_not_sa_pattern = re.compile(r"^(?:sqlalchemy\.(?!testing)|alembic\.)")

2041

2042

2043def _warnings_warn(

2044 message: Union[str, Warning],

2045 category: Optional[Type[Warning]] = None,

2046 stacklevel: int = 2,

2047) -> None:

2048

2049 if category is None and isinstance(message, Warning):

2050 category = type(message)

2051

2052 # adjust the given stacklevel to be outside of SQLAlchemy

2053 try:

2054 frame = sys._getframe(stacklevel)

2055 except ValueError:

2056 # being called from less than 3 (or given) stacklevels, weird,

2057 # but don't crash

2058 stacklevel = 0

2059 except:

2060 # _getframe() doesn't work, weird interpreter issue, weird,

2061 # ok, but don't crash

2062 stacklevel = 0

2063 else:

2064 stacklevel_found = warning_tag_found = False

2065 while frame is not None:

2066 # using __name__ here requires that we have __name__ in the

2067 # __globals__ of the decorated string functions we make also.

2068 # we generate this using {"__name__": fn.__module__}

2069 if not stacklevel_found and not re.match(

2070 _not_sa_pattern, frame.f_globals.get("__name__", "")

2071 ):

2072 # stop incrementing stack level if an out-of-SQLA line

2073 # were found.

2074 stacklevel_found = True

2075

2076 # however, for the warning tag thing, we have to keep

2077 # scanning up the whole traceback

2078

2079 if frame.f_code in _warning_tags:

2080 warning_tag_found = True

2081 _suffix, _category = _warning_tags[frame.f_code]

2082 category = category or _category

2083 message = f"{message} ({_suffix})"

2084

2085 frame = frame.f_back # type: ignore[assignment]

2086

2087 if not stacklevel_found:

2088 stacklevel += 1

2089 elif stacklevel_found and warning_tag_found:

2090 break

2091

2092 if category is not None:

2093 warnings.warn(message, category, stacklevel=stacklevel + 1)

2094 else:

2095 warnings.warn(message, stacklevel=stacklevel + 1)

2096

2097

2098def only_once(

2099 fn: Callable[..., _T], retry_on_exception: bool

2100) -> Callable[..., Optional[_T]]:

2101 """Decorate the given function to be a no-op after it is called exactly

2102 once."""

2103

2104 once = [fn]

2105

2106 def go(*arg: Any, **kw: Any) -> Optional[_T]:

2107 # strong reference fn so that it isn't garbage collected,

2108 # which interferes with the event system's expectations

2109 strong_fn = fn # noqa

2110 if once:

2111 once_fn = once.pop()

2112 try:

2113 return once_fn(*arg, **kw)

2114 except:

2115 if retry_on_exception:

2116 once.insert(0, once_fn)

2117 raise

2119 return None

2121 return go

2124_SQLA_RE = re.compile(r"sqlalchemy/([a-z_]+/){0,2}[a-z_]+\.py")

2125_UNITTEST_RE = re.compile(r"unit(?:2|test2?/)")

2126

2127

2128def chop_traceback(

2129 tb: List[str],

2130 exclude_prefix: re.Pattern[str] = _UNITTEST_RE,

2131 exclude_suffix: re.Pattern[str] = _SQLA_RE,

2132) -> List[str]:

2133 """Chop extraneous lines off beginning and end of a traceback.

2134

2135 :param tb:

2136 a list of traceback lines as returned by ``traceback.format_stack()``

2137

2138 :param exclude_prefix:

2139 a regular expression object matching lines to skip at beginning of

2140 ``tb``

2141

2142 :param exclude_suffix:

2143 a regular expression object matching lines to skip at end of ``tb``

2144 """

2145 start = 0

2146 end = len(tb) - 1

2147 while start <= end and exclude_prefix.search(tb[start]):

2148 start += 1

2149 while start <= end and exclude_suffix.search(tb[end]):

2150 end -= 1

2151 return tb[start : end + 1]

2152

2153

2154def attrsetter(attrname):

2155 code = "def set(obj, value): obj.%s = value" % attrname

2156 env = locals().copy()

2157 exec(code, env)

2158 return env["set"]

2159

2160

2161dunders_re = re.compile("^__.+__$")

2162

2163

2164class TypingOnly:

2165 """A mixin class that marks a class as 'typing only', meaning it has

2166 absolutely no methods, attributes, or runtime functionality whatsoever.

2167

2168 """

2169

2170 __slots__ = ()

2171

2172 def __init_subclass__(cls, **kw: Any) -> None:

2173 if TypingOnly in cls.__bases__:

2174 remaining = {

2175 name for name in cls.__dict__ if not dunders_re.match(name)

2176 }

2177 if remaining:

2178 raise AssertionError(

2179 f"Class {cls} directly inherits TypingOnly but has "

2180 f"additional attributes {remaining}."

2181 )

2182 super().__init_subclass__(**kw)

2183

2184

2185class EnsureKWArg:

2186 r"""Apply translation of functions to accept \**kw arguments if they

2187 don't already.

2188

2189 Used to ensure cross-compatibility with third party legacy code, for things

2190 like compiler visit methods that need to accept ``**kw`` arguments,

2191 but may have been copied from old code that didn't accept them.

2192

2193 """

2194

2195 ensure_kwarg: str

2196 """a regular expression that indicates method names for which the method

2197 should accept ``**kw`` arguments.

2198

2199 The class will scan for methods matching the name template and decorate

2200 them if necessary to ensure ``**kw`` parameters are accepted.

2201

2202 """

2203

2204 def __init_subclass__(cls) -> None:

2205 fn_reg = cls.ensure_kwarg

2206 clsdict = cls.__dict__

2207 if fn_reg:

2208 for key in clsdict:

2209 m = re.match(fn_reg, key)

2210 if m:

2211 fn = clsdict[key]

2212 spec = compat.inspect_getfullargspec(fn)

2213 if not spec.varkw:

2214 wrapped = cls._wrap_w_kw(fn)

2215 setattr(cls, key, wrapped)

2216 super().__init_subclass__()

2217

2218 @classmethod

2219 def _wrap_w_kw(cls, fn: Callable[..., Any]) -> Callable[..., Any]:

2220 def wrap(*arg: Any, **kw: Any) -> Any:

2221 return fn(*arg)

2222

2223 return update_wrapper(wrap, fn)

2224

2225

2226def wrap_callable(wrapper, fn):

2227 """Augment functools.update_wrapper() to work with objects with

2228 a ``__call__()`` method.

2229

2230 :param fn:

2231 object with __call__ method

2232

2233 """

2234 if hasattr(fn, "__name__"):

2235 return update_wrapper(wrapper, fn)

2236 else:

2237 _f = wrapper

2238 _f.__name__ = fn.__class__.__name__

2239 if hasattr(fn, "__module__"):

2240 _f.__module__ = fn.__module__

2241

2242 if hasattr(fn.__call__, "__doc__") and fn.__call__.__doc__:

2243 _f.__doc__ = fn.__call__.__doc__

2244 elif fn.__doc__:

2245 _f.__doc__ = fn.__doc__

2246

2247 return _f

2248

2249

2250def find_matching_paren(text: str, start: int = 0) -> Optional[int]:

2251 """Return the index of the ``)`` that matches the ``(`` at ``start``.

2252

2253 The walk skips single-quoted (``'...'``) and double-quoted (``"..."``)

2254 string literals, so parentheses inside string literals do not affect

2255 the depth counter. ``''`` and ``""`` are treated as escaped quotes

2256 inside their respective contexts, matching PostgreSQL/SQLite literal

2257 conventions.

2258

2259 Returns ``None`` if the opening parenthesis is never closed

2260 (unbalanced). The character at ``text[start]`` must be ``(``.

2261

2262 Note for SQLite use, SQLite also supports MySQL backtick-style quotes as

2263 well as SQL Server bracket style quotes; the latter has different escaping

2264 behaviors. A follow-up patch could add support for these two additional

2265 styles (consider using an enum like QuotingStyle.DOUBLE |

2266 QuotingStyle.BRACKET, etc.)

2267

2268 E.g.::

2269

2270 >>> find_matching_paren("(a + b)")

2271 6

2272 >>> find_matching_paren("((a)(b))")

2273 7

2274 >>> find_matching_paren("(a = '(' AND b = ')')")

2275 20

2276

2277 """

2278 assert text[start] == "(", "start index must point at an open paren"

2279

2280 depth = 0

2281 in_single = False

2282 in_double = False

2283 n = len(text)

2284 i = start

2285 while i < n:

2286 ch = text[i]

2287 if in_single:

2288 if ch == "'":

2289 if i + 1 < n and text[i + 1] == "'":

2290 i += 2

2291 continue

2292 in_single = False

2293 elif in_double:

2294 if ch == '"':

2295 if i + 1 < n and text[i + 1] == '"':

2296 i += 2

2297 continue

2298 in_double = False

2299 elif ch == "'":

2300 in_single = True

2301 elif ch == '"':

2302 in_double = True

2303 elif ch == "(":

2304 depth += 1

2305 elif ch == ")":

2306 depth -= 1

2307 if depth == 0:

2308 return i

2309 i += 1

2310 return None

2311

2312

2313def strip_outer_parens(text: str) -> str:

2314 """Remove one layer of outer parentheses from ``text`` if they wrap the

2315 entire (stripped) string.

2316

2317 Whitespace is preserved if the parentheses do not wrap the whole

2318 expression. String literals are honored via :func:`find_matching_paren`,

2319 so ``"(a = '(' AND b = ')')"`` correctly strips to

2320 ``"a = '(' AND b = ')'"`` rather than being interpreted as two separate

2321 paren groups.

2322

2323 E.g.::

2324

2325 >>> strip_outer_parens("(a IS NOT NULL)")

2326 'a IS NOT NULL'

2327 >>> strip_outer_parens("(a) AND (b)")

2328 '(a) AND (b)'

2329 >>> strip_outer_parens("a NOT NULL")

2330 'a NOT NULL'

2331

2332 """

2333 stripped = text.strip()

2334 lstripped = len(stripped)

2335 if lstripped < 2 or stripped[0] != "(" or stripped[-1] != ")":

2336 return text

2337 close = find_matching_paren(stripped, 0)

2338 if close is not None and close == lstripped - 1:

2339 return stripped[1:-1]

2340 return text

2341

2342

2343def quoted_token_parser(value):

2344 """Parse a dotted identifier with accommodation for quoted names.

2345

2346 Includes support for SQL-style double quotes as a literal character.

2347

2348 E.g.::

2349

2350 >>> quoted_token_parser("name")

2351 ["name"]

2352 >>> quoted_token_parser("schema.name")

2353 ["schema", "name"]

2354 >>> quoted_token_parser('"Schema"."Name"')

2355 ['Schema', 'Name']

2356 >>> quoted_token_parser('"Schema"."Name""Foo"')

2357 ['Schema', 'Name""Foo']

2358

2359 """

2360

2361 if '"' not in value:

2362 return value.split(".")

2363

2364 # 0 = outside of quotes

2365 # 1 = inside of quotes

2366 state = 0

2367 result: List[List[str]] = [[]]

2368 idx = 0

2369 lv = len(value)

2370 while idx < lv:

2371 char = value[idx]

2372 if char == '"':

2373 if state == 1 and idx < lv - 1 and value[idx + 1] == '"':

2374 result[-1].append('"')

2375 idx += 1

2376 else:

2377 state ^= 1

2378 elif char == "." and state == 0:

2379 result.append([])

2380 else:

2381 result[-1].append(char)

2382 idx += 1

2383

2384 return ["".join(token) for token in result]

2385

2386

2387def add_parameter_text(params: Any, text: str) -> Callable[[_F], _F]:

2388 params = _collections.to_list(params)

2389

2390 def decorate(fn):

2391 doc = fn.__doc__ is not None and fn.__doc__ or ""

2392 if doc:

2393 doc = inject_param_text(doc, {param: text for param in params})

2394 fn.__doc__ = doc

2395 return fn

2396

2397 return decorate

2398

2399

2400def _dedent_docstring(text: str) -> str:

2401 split_text = text.split("\n", 1)

2402 if len(split_text) == 1:

2403 return text

2404 else:

2405 firstline, remaining = split_text

2406 if not firstline.startswith(" "):

2407 return firstline + "\n" + textwrap.dedent(remaining)

2408 else:

2409 return textwrap.dedent(text)

2410

2411

2412def inject_docstring_text(

2413 given_doctext: Optional[str], injecttext: str, pos: int

2414) -> str:

2415 doctext: str = _dedent_docstring(given_doctext or "")

2416 lines = doctext.split("\n")

2417 if len(lines) == 1:

2418 lines.append("")

2419 injectlines = textwrap.dedent(injecttext).split("\n")

2420 if injectlines[0]:

2421 injectlines.insert(0, "")

2422

2423 blanks = [num for num, line in enumerate(lines) if not line.strip()]

2424 blanks.insert(0, 0)

2425

2426 inject_pos = blanks[min(pos, len(blanks) - 1)]

2427

2428 lines = lines[0:inject_pos] + injectlines + lines[inject_pos:]

2429 return "\n".join(lines)

2430

2431

2432_param_reg = re.compile(r"(\s+):param (.+?):")

2433

2434

2435def inject_param_text(doctext: str, inject_params: Dict[str, str]) -> str:

2436 doclines = collections.deque(doctext.splitlines())

2437 lines = []

2438

2439 # TODO: this is not working for params like ":param case_sensitive=True:"

2440

2441 to_inject = None

2442 while doclines:

2443 line = doclines.popleft()

2444

2445 m = _param_reg.match(line)

2446

2447 if to_inject is None:

2448 if m:

2449 param = m.group(2).lstrip("*")

2450 if param in inject_params:

2451 # default indent to that of :param: plus one

2452 indent = " " * len(m.group(1)) + " "

2453

2454 # but if the next line has text, use that line's

2455 # indentation

2456 if doclines:

2457 m2 = re.match(r"(\s+)\S", doclines[0])

2458 if m2:

2459 indent = " " * len(m2.group(1))

2460

2461 to_inject = indent + inject_params[param]

2462 elif m:

2463 lines.extend(["\n", to_inject, "\n"])

2464 to_inject = None

2465 elif not line.rstrip():

2466 lines.extend([line, to_inject, "\n"])

2467 to_inject = None

2468 elif line.endswith("::"):

2469 # TODO: this still won't cover if the code example itself has

2470 # blank lines in it, need to detect those via indentation.

2471 lines.extend([line, doclines.popleft()])

2472 continue

2473 lines.append(line)

2474

2475 return "\n".join(lines)

2476

2477

2478def repr_tuple_names(names: List[str]) -> Optional[str]:

2479 """Trims a list of strings from the middle and return a string of up to

2480 four elements. Strings greater than 11 characters will be truncated"""

2481 if len(names) == 0:

2482 return None

2483 flag = len(names) <= 4

2484 names = names[0:4] if flag else names[0:3] + names[-1:]

2485 res = ["%s.." % name[:11] if len(name) > 11 else name for name in names]

2486 if flag:

2487 return ", ".join(res)

2488 else:

2489 return "%s, ..., %s" % (", ".join(res[0:3]), res[-1])

2490

2491

2492def has_compiled_ext(raise_=False):

2493 from ._has_cython import HAS_CYEXTENSION

2494

2495 if HAS_CYEXTENSION:

2496 return True

2497 elif raise_:

2498 raise ImportError(

2499 "cython extensions were expected to be installed, "

2500 "but are not present"

2501 )

2502 else:

2503 return False

2504

2505

2506def load_uncompiled_module(module: _M) -> _M:

2507 """Load the non-compied version of a module that is also

2508 compiled with cython.

2509 """

2510 full_name = module.__name__

2511 assert module.__spec__

2512 parent_name = module.__spec__.parent

2513 assert parent_name

2514 parent_module = sys.modules[parent_name]

2515 assert parent_module.__spec__

2516 package_path = parent_module.__spec__.origin

2517 assert package_path and package_path.endswith("__init__.py")

2518

2519 name = full_name.split(".")[-1]

2520 module_path = package_path.replace("__init__.py", f"{name}.py")

2521

2522 py_spec = importlib.util.spec_from_file_location(full_name, module_path)

2523 assert py_spec

2524 py_module = importlib.util.module_from_spec(py_spec)

2525 assert py_spec.loader

2526 py_spec.loader.exec_module(py_module)

2527 return cast(_M, py_module)

2528

2529

2530_pre_release_normalize = {

2531 "a": "a",

2532 "alpha": "a",

2533 "b": "b",

2534 "beta": "b",

2535 "c": "rc",

2536 "pre": "rc",

2537 "preview": "rc",

2538 "rc": "rc",

2539}

2540

2541_version_string_re = re.compile(

2542 r"""

2543 \s*

2544 (?:[a-z][a-z0-9]*[-_])? # ignored prefix, "py3-"

2545 v?

2546 (?P<release>\d+(?:\.\d+)*)

2547 (?: # pre-release

2548 [-_.]?

2549 (?P<pre_l>alpha|beta|preview|pre|rc|a|b|c)

2550 [-_.]?

2551 (?P<pre_n>\d+)?

2552 )?

2553 (?: # post-release

2554 [-_.]?

2555 (?P<post_l>post|rev|r)

2556 [-_.]?

2557 (?P<post_n>\d+)?

2558 )?

2559 (?: # developmental release

2560 [-_.]?

2561 (?P<dev_l>dev)

2562 [-_.]?

2563 (?P<dev_n>\d+)?

2564 )?

2565 """,

2566 re.X | re.I,

2567)

2568

2569_VersionSortKey = Tuple[

2570 Tuple[int, ...],

2571 Tuple[int, str, int],

2572 Tuple[int, int],

2573 Tuple[int, int],

2574]

2575

2576

2577def _version_sort_key(

2578 release: Tuple[int, ...],

2579 pre: Optional[Tuple[str, int]],

2580 post: Optional[int],

2581 dev: Optional[int],

2582) -> _VersionSortKey:

2583 if pre is None and post is None and dev is not None:

2584 # a dev release with no other qualifiers precedes every

2585 # pre-release of the same release number

2586 pre_key = (-1, "", 0)

2587 elif pre is None:

2588 pre_key = (1, "", 0)

2589 else:

2590 pre_key = (0, pre[0], pre[1])

2591

2592 return (

2593 release,

2594 pre_key,

2595 (0, 0) if post is None else (1, post),

2596 (1, 0) if dev is None else (0, dev),

2597 )

2598

2599

2600def _version_comparison(

2601 op: Callable[[Any, Any], bool],

2602) -> Callable[[VersionInfo, Any], Any]:

2603 """Build one of :class:`.VersionInfo`'s comparison methods.

2604

2605 Comparison takes place against the sort key rather than the tuple

2606 itself, so that pre-release and similar qualifiers are taken into

2607 account. A plain tuple is interpreted as the release segment of a

2608 final release; anything else is not comparable.

2609

2610 """

2611

2612 def compare(self: VersionInfo, other: Any) -> Any:

2613 if isinstance(other, VersionInfo):

2614 other_key = other._sort_key

2615 elif isinstance(other, tuple):

2616 other_key = _version_sort_key(other, None, None, None)

2617 else:

2618 return NotImplemented

2619 return op(self._sort_key, other_key)

2620

2621 return compare

2622

2623

2624class VersionInfo(Tuple[int, ...]):

2625 """A version number, as a tuple of integers.

2626

2627 :class:`.VersionInfo` is a ``tuple`` subclass consisting of the

2628 numeric "release" segment of a version only, e.g. ``2.0.0rc1``

2629 is the tuple ``(2, 0, 0)``. Ordering however takes any

2630 pre-release, post-release and developmental qualifiers into account

2631 as described by :pep:`440`, so that ``2.0.0rc1`` compares as less than

2632 ``2.0.0``, including when compared against a plain tuple such as

2633 ``(2, 0, 0)``.

2634

2635 Plain tuples are interpreted as final releases when compared against

2636 a :class:`.VersionInfo`.

2637

2638 .. versionadded:: 2.1

2639

2640 """

2641

2642 string: Optional[str]

2643 """the string from which this version was parsed, if any."""

2644

2645 pre: Optional[Tuple[str, int]]

2646 """normalized pre-release qualifier, e.g. ``("rc", 1)``."""

2647

2648 post: Optional[int]

2649 """post-release number, if any."""

2650

2651 dev: Optional[int]

2652 """developmental release number, if any."""

2653

2654 _sort_key: _VersionSortKey

2655

2656 def __new__(

2657 cls,

2658 release: Sequence[int] = (),

2659 *,

2660 string: Optional[str] = None,

2661 pre: Optional[Tuple[str, int]] = None,

2662 post: Optional[int] = None,

2663 dev: Optional[int] = None,

2664 ) -> VersionInfo:

2665 # __new__ is needed as the release segment has to be passed to

2666 # tuple.__new__(); the remaining state is set up in __init__

2667 return tuple.__new__(cls, release)

2668

2669 def __init__(

2670 self,

2671 release: Sequence[int] = (),

2672 *,

2673 string: Optional[str] = None,

2674 pre: Optional[Tuple[str, int]] = None,

2675 post: Optional[int] = None,

2676 dev: Optional[int] = None,

2677 ):

2678 self.string = string

2679 self.pre = pre

2680 self.post = post

2681 self.dev = dev

2682 self._sort_key = _version_sort_key(tuple(self), pre, post, dev)

2683

2684 def __repr__(self) -> str:

2685 if self.string is not None:

2686 return f"VersionInfo({tuple(self)!r}, string={self.string!r})"

2687 else:

2688 return f"VersionInfo({tuple(self)!r})"

2689

2690 def __str__(self) -> str:

2691 if self.string is not None:

2692 return self.string

2693 else:

2694 return ".".join(str(num) for num in self)

2695

2696 # every comparison has to be stated explicitly; ``tuple`` implements

2697 # all six of them, so ``functools.total_ordering`` fills in nothing

2698 # here and the ones left out would silently compare as plain tuples

2699 __eq__ = _version_comparison(operator.eq)

2700 __ne__ = _version_comparison(operator.ne)

2701 __lt__ = _version_comparison(operator.lt)

2702 __le__ = _version_comparison(operator.le)

2703 __gt__ = _version_comparison(operator.gt)

2704 __ge__ = _version_comparison(operator.ge)

2705

2706 def __hash__(self) -> int:

2707 return hash(self._sort_key)

2708

2709

2710def parse_version_string(version: Optional[str]) -> VersionInfo:

2711 """Parse a DBAPI version string into a :class:`.VersionInfo`.

2712

2713 Leading characters that are not part of the version itself are

2714 ignored, as are trailing characters following the version, so that

2715 strings such as ``"py3-4.0.19-beta4"`` and

2716 ``"2.9.10 (dt dec pq3 ext lo64)"`` parse correctly.

2717

2718 An empty :class:`.VersionInfo` is returned if no version number can be

2719 located at all.

2720

2721 Parsing is deliberately more tolerant than that of :pep:`440`, which

2722 the version strings published by DBAPIs frequently do not conform to;

2723 a strict implementation such as that of the ``packaging`` library

2724 rejects each of the above outright.

2725

2726 .. versionadded:: 2.1

2727

2728 """

2729

2730 if not version:

2731 return VersionInfo((), string=version)

2732

2733 m = _version_string_re.match(version)

2734 if m is None:

2735 return VersionInfo((), string=version)

2736

2737 release = tuple(int(x) for x in m.group("release").split("."))

2738

2739 pre_l = m.group("pre_l")

2740 pre: Optional[Tuple[str, int]]

2741 if pre_l is not None:

2742 pre = (

2743 _pre_release_normalize[pre_l.lower()],

2744 int(m.group("pre_n") or 0),

2745 )

2746 else:

2747 pre = None

2748

2749 return VersionInfo(

2750 release,

2751 string=version,

2752 pre=pre,

2753 post=(

2754 int(m.group("post_n") or 0)

2755 if m.group("post_l") is not None

2756 else None

2757 ),

2758 dev=(

2759 int(m.group("dev_n") or 0)

2760 if m.group("dev_l") is not None

2761 else None

2762 ),

2763 )

2764

2765

2766def parse_version_from_metadata(distribution: str) -> VersionInfo:

2767 """Return the version of an installed distribution as a

2768 :class:`.VersionInfo`.

2769

2770 This is intended for use by dialects whose DBAPI module does not

2771 itself publish a version number, such as ``asyncmy``. As the

2772 distribution name is not necessarily the same as the module name, and

2773 the installed distribution is not necessarily the module that was

2774 imported, this should not be used when the DBAPI module provides a

2775 version of its own.

2776

2777 An empty :class:`.VersionInfo` is returned if the distribution is not

2778 installed.

2779

2780 .. versionadded:: 2.1

2781

2782 """

2783

2784 try:

2785 version = importlib.metadata.version(distribution)

2786 except importlib.metadata.PackageNotFoundError:

2787 return VersionInfo()

2788 else:

2789 return parse_version_string(version)

2790

2791

2792class _Missing(enum.Enum):

2793 Missing = enum.auto()

2794

2795

2796Missing = _Missing.Missing

2797MissingOr = Union[_T, Literal[_Missing.Missing]]