Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/util»
This commit is contained in:
322
venv/Lib/site-packages/sqlalchemy/util/queue.py
Normal file
322
venv/Lib/site-packages/sqlalchemy/util/queue.py
Normal file
@@ -0,0 +1,322 @@
|
||||
# util/queue.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# This module is part of SQLAlchemy and is released under
|
||||
# the MIT License: https://www.opensource.org/licenses/mit-license.php
|
||||
# mypy: allow-untyped-defs, allow-untyped-calls
|
||||
|
||||
"""An adaptation of Py2.3/2.4's Queue module which supports reentrant
|
||||
behavior, using RLock instead of Lock for its mutex object. The
|
||||
Queue object is used exclusively by the sqlalchemy.pool.QueuePool
|
||||
class.
|
||||
|
||||
This is to support the connection pool's usage of weakref callbacks to return
|
||||
connections to the underlying Queue, which can in extremely
|
||||
rare cases be invoked within the ``get()`` method of the Queue itself,
|
||||
producing a ``put()`` inside the ``get()`` and therefore a reentrant
|
||||
condition.
|
||||
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections import deque
|
||||
import threading
|
||||
from time import time as _time
|
||||
import typing
|
||||
from typing import Any
|
||||
from typing import Awaitable
|
||||
from typing import Deque
|
||||
from typing import Generic
|
||||
from typing import Optional
|
||||
from typing import TypeVar
|
||||
|
||||
from .concurrency import await_fallback
|
||||
from .concurrency import await_only
|
||||
from .langhelpers import memoized_property
|
||||
|
||||
_T = TypeVar("_T", bound=Any)
|
||||
__all__ = ["Empty", "Full", "Queue"]
|
||||
|
||||
|
||||
class Empty(Exception):
|
||||
"Exception raised by Queue.get(block=0)/get_nowait()."
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class Full(Exception):
|
||||
"Exception raised by Queue.put(block=0)/put_nowait()."
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class QueueCommon(Generic[_T]):
|
||||
maxsize: int
|
||||
use_lifo: bool
|
||||
|
||||
def __init__(self, maxsize: int = 0, use_lifo: bool = False): ...
|
||||
|
||||
def empty(self) -> bool:
|
||||
raise NotImplementedError()
|
||||
|
||||
def full(self) -> bool:
|
||||
raise NotImplementedError()
|
||||
|
||||
def qsize(self) -> int:
|
||||
raise NotImplementedError()
|
||||
|
||||
def put_nowait(self, item: _T) -> None:
|
||||
raise NotImplementedError()
|
||||
|
||||
def put(
|
||||
self, item: _T, block: bool = True, timeout: Optional[float] = None
|
||||
) -> None:
|
||||
raise NotImplementedError()
|
||||
|
||||
def get_nowait(self) -> _T:
|
||||
raise NotImplementedError()
|
||||
|
||||
def get(self, block: bool = True, timeout: Optional[float] = None) -> _T:
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
class Queue(QueueCommon[_T]):
|
||||
queue: Deque[_T]
|
||||
|
||||
def __init__(self, maxsize: int = 0, use_lifo: bool = False):
|
||||
"""Initialize a queue object with a given maximum size.
|
||||
|
||||
If `maxsize` is <= 0, the queue size is infinite.
|
||||
|
||||
If `use_lifo` is True, this Queue acts like a Stack (LIFO).
|
||||
"""
|
||||
|
||||
self._init(maxsize)
|
||||
# mutex must be held whenever the queue is mutating. All methods
|
||||
# that acquire mutex must release it before returning. mutex
|
||||
# is shared between the two conditions, so acquiring and
|
||||
# releasing the conditions also acquires and releases mutex.
|
||||
self.mutex = threading.RLock()
|
||||
# Notify not_empty whenever an item is added to the queue; a
|
||||
# thread waiting to get is notified then.
|
||||
self.not_empty = threading.Condition(self.mutex)
|
||||
# Notify not_full whenever an item is removed from the queue;
|
||||
# a thread waiting to put is notified then.
|
||||
self.not_full = threading.Condition(self.mutex)
|
||||
# If this queue uses LIFO or FIFO
|
||||
self.use_lifo = use_lifo
|
||||
|
||||
def qsize(self) -> int:
|
||||
"""Return the approximate size of the queue (not reliable!)."""
|
||||
|
||||
with self.mutex:
|
||||
return self._qsize()
|
||||
|
||||
def empty(self) -> bool:
|
||||
"""Return True if the queue is empty, False otherwise (not
|
||||
reliable!)."""
|
||||
|
||||
with self.mutex:
|
||||
return self._empty()
|
||||
|
||||
def full(self) -> bool:
|
||||
"""Return True if the queue is full, False otherwise (not
|
||||
reliable!)."""
|
||||
|
||||
with self.mutex:
|
||||
return self._full()
|
||||
|
||||
def put(
|
||||
self, item: _T, block: bool = True, timeout: Optional[float] = None
|
||||
) -> None:
|
||||
"""Put an item into the queue.
|
||||
|
||||
If optional args `block` is True and `timeout` is None (the
|
||||
default), block if necessary until a free slot is
|
||||
available. If `timeout` is a positive number, it blocks at
|
||||
most `timeout` seconds and raises the ``Full`` exception if no
|
||||
free slot was available within that time. Otherwise (`block`
|
||||
is false), put an item on the queue if a free slot is
|
||||
immediately available, else raise the ``Full`` exception
|
||||
(`timeout` is ignored in that case).
|
||||
"""
|
||||
|
||||
with self.not_full:
|
||||
if not block:
|
||||
if self._full():
|
||||
raise Full
|
||||
elif timeout is None:
|
||||
while self._full():
|
||||
self.not_full.wait()
|
||||
else:
|
||||
if timeout < 0:
|
||||
raise ValueError("'timeout' must be a positive number")
|
||||
endtime = _time() + timeout
|
||||
while self._full():
|
||||
remaining = endtime - _time()
|
||||
if remaining <= 0.0:
|
||||
raise Full
|
||||
self.not_full.wait(remaining)
|
||||
self._put(item)
|
||||
self.not_empty.notify()
|
||||
|
||||
def put_nowait(self, item: _T) -> None:
|
||||
"""Put an item into the queue without blocking.
|
||||
|
||||
Only enqueue the item if a free slot is immediately available.
|
||||
Otherwise raise the ``Full`` exception.
|
||||
"""
|
||||
return self.put(item, False)
|
||||
|
||||
def get(self, block: bool = True, timeout: Optional[float] = None) -> _T:
|
||||
"""Remove and return an item from the queue.
|
||||
|
||||
If optional args `block` is True and `timeout` is None (the
|
||||
default), block if necessary until an item is available. If
|
||||
`timeout` is a positive number, it blocks at most `timeout`
|
||||
seconds and raises the ``Empty`` exception if no item was
|
||||
available within that time. Otherwise (`block` is false),
|
||||
return an item if one is immediately available, else raise the
|
||||
``Empty`` exception (`timeout` is ignored in that case).
|
||||
|
||||
"""
|
||||
with self.not_empty:
|
||||
if not block:
|
||||
if self._empty():
|
||||
raise Empty
|
||||
elif timeout is None:
|
||||
while self._empty():
|
||||
self.not_empty.wait()
|
||||
else:
|
||||
if timeout < 0:
|
||||
raise ValueError("'timeout' must be a positive number")
|
||||
endtime = _time() + timeout
|
||||
while self._empty():
|
||||
remaining = endtime - _time()
|
||||
if remaining <= 0.0:
|
||||
raise Empty
|
||||
self.not_empty.wait(remaining)
|
||||
item = self._get()
|
||||
self.not_full.notify()
|
||||
return item
|
||||
|
||||
def get_nowait(self) -> _T:
|
||||
"""Remove and return an item from the queue without blocking.
|
||||
|
||||
Only get an item if one is immediately available. Otherwise
|
||||
raise the ``Empty`` exception.
|
||||
"""
|
||||
|
||||
return self.get(False)
|
||||
|
||||
def _init(self, maxsize: int) -> None:
|
||||
self.maxsize = maxsize
|
||||
self.queue = deque()
|
||||
|
||||
def _qsize(self) -> int:
|
||||
return len(self.queue)
|
||||
|
||||
def _empty(self) -> bool:
|
||||
return not self.queue
|
||||
|
||||
def _full(self) -> bool:
|
||||
return self.maxsize > 0 and len(self.queue) == self.maxsize
|
||||
|
||||
def _put(self, item: _T) -> None:
|
||||
self.queue.append(item)
|
||||
|
||||
def _get(self) -> _T:
|
||||
if self.use_lifo:
|
||||
# LIFO
|
||||
return self.queue.pop()
|
||||
else:
|
||||
# FIFO
|
||||
return self.queue.popleft()
|
||||
|
||||
|
||||
class AsyncAdaptedQueue(QueueCommon[_T]):
|
||||
if typing.TYPE_CHECKING:
|
||||
|
||||
@staticmethod
|
||||
def await_(coroutine: Awaitable[Any]) -> _T: ...
|
||||
|
||||
else:
|
||||
await_ = staticmethod(await_only)
|
||||
|
||||
def __init__(self, maxsize: int = 0, use_lifo: bool = False):
|
||||
self.use_lifo = use_lifo
|
||||
self.maxsize = maxsize
|
||||
|
||||
def empty(self) -> bool:
|
||||
return self._queue.empty()
|
||||
|
||||
def full(self):
|
||||
return self._queue.full()
|
||||
|
||||
def qsize(self):
|
||||
return self._queue.qsize()
|
||||
|
||||
@memoized_property
|
||||
def _queue(self) -> asyncio.Queue[_T]:
|
||||
# Delay creation of the queue until it is first used, to avoid
|
||||
# binding it to a possibly wrong event loop.
|
||||
# By delaying the creation of the pool we accommodate the common
|
||||
# usage pattern of instantiating the engine at module level, where a
|
||||
# different event loop is in present compared to when the application
|
||||
# is actually run.
|
||||
|
||||
queue: asyncio.Queue[_T]
|
||||
|
||||
if self.use_lifo:
|
||||
queue = asyncio.LifoQueue(maxsize=self.maxsize)
|
||||
else:
|
||||
queue = asyncio.Queue(maxsize=self.maxsize)
|
||||
return queue
|
||||
|
||||
def put_nowait(self, item: _T) -> None:
|
||||
try:
|
||||
self._queue.put_nowait(item)
|
||||
except asyncio.QueueFull as err:
|
||||
raise Full() from err
|
||||
|
||||
def put(
|
||||
self, item: _T, block: bool = True, timeout: Optional[float] = None
|
||||
) -> None:
|
||||
if not block:
|
||||
return self.put_nowait(item)
|
||||
|
||||
try:
|
||||
if timeout is not None:
|
||||
self.await_(asyncio.wait_for(self._queue.put(item), timeout))
|
||||
else:
|
||||
self.await_(self._queue.put(item))
|
||||
except (asyncio.QueueFull, asyncio.TimeoutError) as err:
|
||||
raise Full() from err
|
||||
|
||||
def get_nowait(self) -> _T:
|
||||
try:
|
||||
return self._queue.get_nowait()
|
||||
except asyncio.QueueEmpty as err:
|
||||
raise Empty() from err
|
||||
|
||||
def get(self, block: bool = True, timeout: Optional[float] = None) -> _T:
|
||||
if not block:
|
||||
return self.get_nowait()
|
||||
|
||||
try:
|
||||
if timeout is not None:
|
||||
return self.await_(
|
||||
asyncio.wait_for(self._queue.get(), timeout)
|
||||
)
|
||||
else:
|
||||
return self.await_(self._queue.get())
|
||||
except (asyncio.QueueEmpty, asyncio.TimeoutError) as err:
|
||||
raise Empty() from err
|
||||
|
||||
|
||||
class FallbackAsyncAdaptedQueue(AsyncAdaptedQueue[_T]):
|
||||
if not typing.TYPE_CHECKING:
|
||||
await_ = staticmethod(await_fallback)
|
||||
202
venv/Lib/site-packages/sqlalchemy/util/tool_support.py
Normal file
202
venv/Lib/site-packages/sqlalchemy/util/tool_support.py
Normal file
@@ -0,0 +1,202 @@
|
||||
# util/tool_support.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# This module is part of SQLAlchemy and is released under
|
||||
# the MIT License: https://www.opensource.org/licenses/mit-license.php
|
||||
# mypy: allow-untyped-defs, allow-untyped-calls
|
||||
"""support routines for the helpers in tools/.
|
||||
|
||||
These aren't imported by the enclosing util package as the are not
|
||||
needed for normal library use.
|
||||
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from argparse import ArgumentParser
|
||||
from argparse import Namespace
|
||||
import contextlib
|
||||
import difflib
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shlex
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import Any
|
||||
from typing import Dict
|
||||
from typing import Iterator
|
||||
from typing import Optional
|
||||
from typing import Union
|
||||
|
||||
from . import compat
|
||||
|
||||
|
||||
class code_writer_cmd:
|
||||
parser: ArgumentParser
|
||||
args: Namespace
|
||||
suppress_output: bool
|
||||
diffs_detected: bool
|
||||
source_root: Path
|
||||
pyproject_toml_path: Path
|
||||
|
||||
def __init__(self, tool_script: str):
|
||||
self.source_root = Path(tool_script).parent.parent
|
||||
self.pyproject_toml_path = self.source_root / Path("pyproject.toml")
|
||||
assert self.pyproject_toml_path.exists()
|
||||
|
||||
self.parser = ArgumentParser()
|
||||
self.parser.add_argument(
|
||||
"--stdout",
|
||||
action="store_true",
|
||||
help="Write to stdout instead of saving to file",
|
||||
)
|
||||
self.parser.add_argument(
|
||||
"-c",
|
||||
"--check",
|
||||
help="Don't write the files back, just return the "
|
||||
"status. Return code 0 means nothing would change. "
|
||||
"Return code 1 means some files would be reformatted",
|
||||
action="store_true",
|
||||
)
|
||||
|
||||
def run_zimports(self, tempfile: str) -> None:
|
||||
self._run_console_script(
|
||||
str(tempfile),
|
||||
{
|
||||
"entrypoint": "zimports",
|
||||
"options": f"--toml-config {self.pyproject_toml_path}",
|
||||
},
|
||||
)
|
||||
|
||||
def run_black(self, tempfile: str) -> None:
|
||||
self._run_console_script(
|
||||
str(tempfile),
|
||||
{
|
||||
"entrypoint": "black",
|
||||
"options": f"--config {self.pyproject_toml_path}",
|
||||
},
|
||||
)
|
||||
|
||||
def _run_console_script(self, path: str, options: Dict[str, Any]) -> None:
|
||||
"""Run a Python console application from within the process.
|
||||
|
||||
Used for black, zimports
|
||||
|
||||
"""
|
||||
|
||||
is_posix = os.name == "posix"
|
||||
|
||||
entrypoint_name = options["entrypoint"]
|
||||
|
||||
for entry in compat.importlib_metadata_get("console_scripts"):
|
||||
if entry.name == entrypoint_name:
|
||||
impl = entry
|
||||
break
|
||||
else:
|
||||
raise Exception(
|
||||
f"Could not find entrypoint console_scripts.{entrypoint_name}"
|
||||
)
|
||||
cmdline_options_str = options.get("options", "")
|
||||
cmdline_options_list = shlex.split(
|
||||
cmdline_options_str, posix=is_posix
|
||||
) + [path]
|
||||
|
||||
kw: Dict[str, Any] = {}
|
||||
if self.suppress_output:
|
||||
kw["stdout"] = kw["stderr"] = subprocess.DEVNULL
|
||||
|
||||
subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-c",
|
||||
"import %s; %s.%s()" % (impl.module, impl.module, impl.attr),
|
||||
]
|
||||
+ cmdline_options_list,
|
||||
cwd=str(self.source_root),
|
||||
**kw,
|
||||
)
|
||||
|
||||
def write_status(self, *text: str) -> None:
|
||||
if not self.suppress_output:
|
||||
sys.stderr.write(" ".join(text))
|
||||
|
||||
def write_output_file_from_text(
|
||||
self, text: str, destination_path: Union[str, Path]
|
||||
) -> None:
|
||||
if self.args.check:
|
||||
self._run_diff(destination_path, source=text)
|
||||
elif self.args.stdout:
|
||||
print(text)
|
||||
else:
|
||||
self.write_status(f"Writing {destination_path}...")
|
||||
Path(destination_path).write_text(
|
||||
text, encoding="utf-8", newline="\n"
|
||||
)
|
||||
self.write_status("done\n")
|
||||
|
||||
def write_output_file_from_tempfile(
|
||||
self, tempfile: str, destination_path: str
|
||||
) -> None:
|
||||
if self.args.check:
|
||||
self._run_diff(destination_path, source_file=tempfile)
|
||||
os.unlink(tempfile)
|
||||
elif self.args.stdout:
|
||||
with open(tempfile) as tf:
|
||||
print(tf.read())
|
||||
os.unlink(tempfile)
|
||||
else:
|
||||
self.write_status(f"Writing {destination_path}...")
|
||||
shutil.move(tempfile, destination_path)
|
||||
self.write_status("done\n")
|
||||
|
||||
def _run_diff(
|
||||
self,
|
||||
destination_path: Union[str, Path],
|
||||
*,
|
||||
source: Optional[str] = None,
|
||||
source_file: Optional[str] = None,
|
||||
) -> None:
|
||||
if source_file:
|
||||
with open(source_file, encoding="utf-8") as tf:
|
||||
source_lines = list(tf)
|
||||
elif source is not None:
|
||||
source_lines = source.splitlines(keepends=True)
|
||||
else:
|
||||
assert False, "source or source_file is required"
|
||||
|
||||
with open(destination_path, encoding="utf-8") as dp:
|
||||
d = difflib.unified_diff(
|
||||
list(dp),
|
||||
source_lines,
|
||||
fromfile=Path(destination_path).as_posix(),
|
||||
tofile="<proposed changes>",
|
||||
n=3,
|
||||
lineterm="\n",
|
||||
)
|
||||
d_as_list = list(d)
|
||||
if d_as_list:
|
||||
self.diffs_detected = True
|
||||
print("".join(d_as_list))
|
||||
|
||||
@contextlib.contextmanager
|
||||
def add_arguments(self) -> Iterator[ArgumentParser]:
|
||||
yield self.parser
|
||||
|
||||
@contextlib.contextmanager
|
||||
def run_program(self) -> Iterator[None]:
|
||||
self.args = self.parser.parse_args()
|
||||
if self.args.check:
|
||||
self.diffs_detected = False
|
||||
self.suppress_output = True
|
||||
elif self.args.stdout:
|
||||
self.suppress_output = True
|
||||
else:
|
||||
self.suppress_output = False
|
||||
yield
|
||||
|
||||
if self.args.check and self.diffs_detected:
|
||||
sys.exit(1)
|
||||
else:
|
||||
sys.exit(0)
|
||||
120
venv/Lib/site-packages/sqlalchemy/util/topological.py
Normal file
120
venv/Lib/site-packages/sqlalchemy/util/topological.py
Normal file
@@ -0,0 +1,120 @@
|
||||
# util/topological.py
|
||||
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# This module is part of SQLAlchemy and is released under
|
||||
# the MIT License: https://www.opensource.org/licenses/mit-license.php
|
||||
|
||||
"""Topological sorting algorithms."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from typing import Collection
|
||||
from typing import DefaultDict
|
||||
from typing import Iterable
|
||||
from typing import Iterator
|
||||
from typing import Sequence
|
||||
from typing import Set
|
||||
from typing import Tuple
|
||||
from typing import TypeVar
|
||||
|
||||
from .. import util
|
||||
from ..exc import CircularDependencyError
|
||||
|
||||
_T = TypeVar("_T", bound=Any)
|
||||
|
||||
__all__ = ["sort", "sort_as_subsets", "find_cycles"]
|
||||
|
||||
|
||||
def sort_as_subsets(
|
||||
tuples: Collection[Tuple[_T, _T]], allitems: Collection[_T]
|
||||
) -> Iterator[Sequence[_T]]:
|
||||
edges: DefaultDict[_T, Set[_T]] = util.defaultdict(set)
|
||||
for parent, child in tuples:
|
||||
edges[child].add(parent)
|
||||
|
||||
todo = list(allitems)
|
||||
todo_set = set(allitems)
|
||||
|
||||
while todo_set:
|
||||
output = []
|
||||
for node in todo:
|
||||
if todo_set.isdisjoint(edges[node]):
|
||||
output.append(node)
|
||||
|
||||
if not output:
|
||||
raise CircularDependencyError(
|
||||
"Circular dependency detected.",
|
||||
find_cycles(tuples, allitems),
|
||||
_gen_edges(edges),
|
||||
)
|
||||
|
||||
todo_set.difference_update(output)
|
||||
todo = [t for t in todo if t in todo_set]
|
||||
yield output
|
||||
|
||||
|
||||
def sort(
|
||||
tuples: Collection[Tuple[_T, _T]],
|
||||
allitems: Collection[_T],
|
||||
deterministic_order: bool = True,
|
||||
) -> Iterator[_T]:
|
||||
"""sort the given list of items by dependency.
|
||||
|
||||
'tuples' is a list of tuples representing a partial ordering.
|
||||
|
||||
deterministic_order is no longer used, the order is now always
|
||||
deterministic given the order of "allitems". the flag is there
|
||||
for backwards compatibility with Alembic.
|
||||
|
||||
"""
|
||||
|
||||
for set_ in sort_as_subsets(tuples, allitems):
|
||||
yield from set_
|
||||
|
||||
|
||||
def find_cycles(
|
||||
tuples: Iterable[Tuple[_T, _T]], allitems: Iterable[_T]
|
||||
) -> Set[_T]:
|
||||
# adapted from:
|
||||
# https://neopythonic.blogspot.com/2009/01/detecting-cycles-in-directed-graph.html
|
||||
|
||||
edges: DefaultDict[_T, Set[_T]] = util.defaultdict(set)
|
||||
for parent, child in tuples:
|
||||
edges[parent].add(child)
|
||||
nodes_to_test = set(edges)
|
||||
|
||||
output = set()
|
||||
|
||||
# we'd like to find all nodes that are
|
||||
# involved in cycles, so we do the full
|
||||
# pass through the whole thing for each
|
||||
# node in the original list.
|
||||
|
||||
# we can go just through parent edge nodes.
|
||||
# if a node is only a child and never a parent,
|
||||
# by definition it can't be part of a cycle. same
|
||||
# if it's not in the edges at all.
|
||||
for node in nodes_to_test:
|
||||
stack = [node]
|
||||
todo = nodes_to_test.difference(stack)
|
||||
while stack:
|
||||
top = stack[-1]
|
||||
for node in edges[top]:
|
||||
if node in stack:
|
||||
cyc = stack[stack.index(node) :]
|
||||
todo.difference_update(cyc)
|
||||
output.update(cyc)
|
||||
|
||||
if node in todo:
|
||||
stack.append(node)
|
||||
todo.remove(node)
|
||||
break
|
||||
else:
|
||||
stack.pop()
|
||||
return output
|
||||
|
||||
|
||||
def _gen_edges(edges: DefaultDict[_T, Set[_T]]) -> Set[Tuple[_T, _T]]:
|
||||
return {(right, left) for left in edges for right in edges[left]}
|
||||
734
venv/Lib/site-packages/sqlalchemy/util/typing.py
Normal file
734
venv/Lib/site-packages/sqlalchemy/util/typing.py
Normal file
@@ -0,0 +1,734 @@
|
||||
# util/typing.py
|
||||
# Copyright (C) 2022-2026 the SQLAlchemy authors and contributors
|
||||
# <see AUTHORS file>
|
||||
#
|
||||
# This module is part of SQLAlchemy and is released under
|
||||
# the MIT License: https://www.opensource.org/licenses/mit-license.php
|
||||
# mypy: allow-untyped-defs, allow-untyped-calls
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import builtins
|
||||
from collections import deque
|
||||
import collections.abc as collections_abc
|
||||
import re
|
||||
import sys
|
||||
import typing
|
||||
from typing import Any
|
||||
from typing import Callable
|
||||
from typing import Dict
|
||||
from typing import ForwardRef
|
||||
from typing import Generic
|
||||
from typing import Iterable
|
||||
from typing import Mapping
|
||||
from typing import NewType
|
||||
from typing import NoReturn
|
||||
from typing import Optional
|
||||
from typing import overload
|
||||
from typing import Set
|
||||
from typing import Tuple
|
||||
from typing import Type
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import TypeVar
|
||||
from typing import Union
|
||||
|
||||
import typing_extensions
|
||||
|
||||
from . import compat
|
||||
|
||||
if True: # zimports removes the tailing comments
|
||||
from typing_extensions import Annotated as Annotated # 3.8
|
||||
from typing_extensions import Concatenate as Concatenate # 3.10
|
||||
from typing_extensions import (
|
||||
dataclass_transform as dataclass_transform, # 3.11,
|
||||
)
|
||||
from typing_extensions import Final as Final # 3.8
|
||||
from typing_extensions import final as final # 3.8
|
||||
from typing_extensions import get_args as get_args # 3.10
|
||||
from typing_extensions import get_origin as get_origin # 3.10
|
||||
from typing_extensions import Literal as Literal # 3.8
|
||||
from typing_extensions import NotRequired as NotRequired # 3.11
|
||||
from typing_extensions import ParamSpec as ParamSpec # 3.10
|
||||
from typing_extensions import Protocol as Protocol # 3.8
|
||||
from typing_extensions import SupportsIndex as SupportsIndex # 3.8
|
||||
from typing_extensions import TypeAlias as TypeAlias # 3.10
|
||||
from typing_extensions import TypedDict as TypedDict # 3.8
|
||||
from typing_extensions import TypeGuard as TypeGuard # 3.10
|
||||
from typing_extensions import Self as Self # 3.11
|
||||
from typing_extensions import TypeAliasType as TypeAliasType # 3.12
|
||||
from typing_extensions import Never as Never # 3.11
|
||||
from typing_extensions import LiteralString as LiteralString # 3.11
|
||||
|
||||
_T = TypeVar("_T", bound=Any)
|
||||
_KT = TypeVar("_KT")
|
||||
_KT_co = TypeVar("_KT_co", covariant=True)
|
||||
_KT_contra = TypeVar("_KT_contra", contravariant=True)
|
||||
_VT = TypeVar("_VT")
|
||||
_VT_co = TypeVar("_VT_co", covariant=True)
|
||||
|
||||
if compat.py310:
|
||||
# why they took until py310 to put this in stdlib is beyond me,
|
||||
# I've been wanting it since py27
|
||||
from types import NoneType as NoneType
|
||||
else:
|
||||
NoneType = type(None) # type: ignore
|
||||
|
||||
|
||||
def is_fwd_none(typ: Any) -> bool:
|
||||
return isinstance(typ, ForwardRef) and typ.__forward_arg__ == "None"
|
||||
|
||||
|
||||
_AnnotationScanType = Union[
|
||||
Type[Any], str, ForwardRef, NewType, TypeAliasType, "GenericProtocol[Any]"
|
||||
]
|
||||
|
||||
|
||||
class ArgsTypeProtocol(Protocol):
|
||||
"""protocol for types that have ``__args__``
|
||||
|
||||
there's no public interface for this AFAIK
|
||||
|
||||
"""
|
||||
|
||||
__args__: Tuple[_AnnotationScanType, ...]
|
||||
|
||||
|
||||
class GenericProtocol(Protocol[_T]):
|
||||
"""protocol for generic types.
|
||||
|
||||
this since Python.typing _GenericAlias is private
|
||||
|
||||
"""
|
||||
|
||||
__args__: Tuple[_AnnotationScanType, ...]
|
||||
__origin__: Type[_T]
|
||||
|
||||
# Python's builtin _GenericAlias has this method, however builtins like
|
||||
# list, dict, etc. do not, even though they have ``__origin__`` and
|
||||
# ``__args__``
|
||||
#
|
||||
# def copy_with(self, params: Tuple[_AnnotationScanType, ...]) -> Type[_T]:
|
||||
# ...
|
||||
|
||||
|
||||
# copied from TypeShed, required in order to implement
|
||||
# MutableMapping.update()
|
||||
class SupportsKeysAndGetItem(Protocol[_KT, _VT_co]):
|
||||
def keys(self) -> Iterable[_KT]: ...
|
||||
|
||||
def __getitem__(self, __k: _KT) -> _VT_co: ...
|
||||
|
||||
|
||||
# work around https://github.com/microsoft/pyright/issues/3025
|
||||
_LiteralStar = Literal["*"]
|
||||
|
||||
|
||||
def de_stringify_annotation(
|
||||
cls: Type[Any],
|
||||
annotation: _AnnotationScanType,
|
||||
originating_module: str,
|
||||
locals_: Mapping[str, Any],
|
||||
*,
|
||||
str_cleanup_fn: Optional[Callable[[str, str], str]] = None,
|
||||
include_generic: bool = False,
|
||||
_already_seen: Optional[Set[Any]] = None,
|
||||
) -> Type[Any]:
|
||||
"""Resolve annotations that may be string based into real objects.
|
||||
|
||||
This is particularly important if a module defines "from __future__ import
|
||||
annotations", as everything inside of __annotations__ is a string. We want
|
||||
to at least have generic containers like ``Mapped``, ``Union``, ``List``,
|
||||
etc.
|
||||
|
||||
"""
|
||||
# looked at typing.get_type_hints(), looked at pydantic. We need much
|
||||
# less here, and we here try to not use any private typing internals
|
||||
# or construct ForwardRef objects which is documented as something
|
||||
# that should be avoided.
|
||||
|
||||
original_annotation = annotation
|
||||
|
||||
if is_fwd_ref(annotation):
|
||||
annotation = annotation.__forward_arg__
|
||||
|
||||
if isinstance(annotation, str):
|
||||
if str_cleanup_fn:
|
||||
annotation = str_cleanup_fn(annotation, originating_module)
|
||||
|
||||
annotation = eval_expression(
|
||||
annotation, originating_module, locals_=locals_, in_class=cls
|
||||
)
|
||||
|
||||
if (
|
||||
include_generic
|
||||
and is_generic(annotation)
|
||||
and not is_literal(annotation)
|
||||
):
|
||||
if _already_seen is None:
|
||||
_already_seen = set()
|
||||
|
||||
if annotation in _already_seen:
|
||||
# only occurs recursively. outermost return type
|
||||
# will always be Type.
|
||||
# the element here will be either ForwardRef or
|
||||
# Optional[ForwardRef]
|
||||
return original_annotation # type: ignore
|
||||
else:
|
||||
_already_seen.add(annotation)
|
||||
|
||||
elements = tuple(
|
||||
de_stringify_annotation(
|
||||
cls,
|
||||
elem,
|
||||
originating_module,
|
||||
locals_,
|
||||
str_cleanup_fn=str_cleanup_fn,
|
||||
include_generic=include_generic,
|
||||
_already_seen=_already_seen,
|
||||
)
|
||||
for elem in annotation.__args__
|
||||
)
|
||||
|
||||
return _copy_generic_annotation_with(annotation, elements)
|
||||
|
||||
return annotation # type: ignore
|
||||
|
||||
|
||||
def fixup_container_fwd_refs(
|
||||
type_: _AnnotationScanType,
|
||||
) -> _AnnotationScanType:
|
||||
"""Correct dict['x', 'y'] into dict[ForwardRef('x'), ForwardRef('y')]
|
||||
and similar for list, set
|
||||
|
||||
"""
|
||||
|
||||
if (
|
||||
is_generic(type_)
|
||||
and get_origin(type_)
|
||||
in (
|
||||
dict,
|
||||
set,
|
||||
list,
|
||||
collections_abc.MutableSet,
|
||||
collections_abc.MutableMapping,
|
||||
collections_abc.MutableSequence,
|
||||
collections_abc.Mapping,
|
||||
collections_abc.Sequence,
|
||||
)
|
||||
# fight, kick and scream to struggle to tell the difference between
|
||||
# dict[] and typing.Dict[] which DO NOT compare the same and DO NOT
|
||||
# behave the same yet there is NO WAY to distinguish between which type
|
||||
# it is using public attributes
|
||||
and not re.match(
|
||||
"typing.(?:Dict|List|Set|.*Mapping|.*Sequence|.*Set)", repr(type_)
|
||||
)
|
||||
):
|
||||
# compat with py3.10 and earlier
|
||||
return get_origin(type_).__class_getitem__( # type: ignore
|
||||
tuple(
|
||||
[
|
||||
ForwardRef(elem) if isinstance(elem, str) else elem
|
||||
for elem in get_args(type_)
|
||||
]
|
||||
)
|
||||
)
|
||||
return type_
|
||||
|
||||
|
||||
def _copy_generic_annotation_with(
|
||||
annotation: GenericProtocol[_T], elements: Tuple[_AnnotationScanType, ...]
|
||||
) -> Type[_T]:
|
||||
if hasattr(annotation, "copy_with"):
|
||||
# List, Dict, etc. real generics
|
||||
return annotation.copy_with(elements) # type: ignore
|
||||
else:
|
||||
# Python builtins list, dict, etc.
|
||||
return annotation.__origin__[elements] # type: ignore
|
||||
|
||||
|
||||
def eval_expression(
|
||||
expression: str,
|
||||
module_name: str,
|
||||
*,
|
||||
locals_: Optional[Mapping[str, Any]] = None,
|
||||
in_class: Optional[Type[Any]] = None,
|
||||
) -> Any:
|
||||
try:
|
||||
base_globals: Dict[str, Any] = sys.modules[module_name].__dict__
|
||||
except KeyError as ke:
|
||||
raise NameError(
|
||||
f"Module {module_name} isn't present in sys.modules; can't "
|
||||
f"evaluate expression {expression}"
|
||||
) from ke
|
||||
|
||||
try:
|
||||
if in_class is not None:
|
||||
cls_namespace = dict(in_class.__dict__)
|
||||
cls_namespace.setdefault(in_class.__name__, in_class)
|
||||
|
||||
# see #10899. We want the locals/globals to take precedence
|
||||
# over the class namespace in this context, even though this
|
||||
# is not the usual way variables would resolve.
|
||||
cls_namespace.update(base_globals)
|
||||
|
||||
annotation = eval(expression, cls_namespace, locals_)
|
||||
else:
|
||||
annotation = eval(expression, base_globals, locals_)
|
||||
except Exception as err:
|
||||
raise NameError(
|
||||
f"Could not de-stringify annotation {expression!r}"
|
||||
) from err
|
||||
else:
|
||||
return annotation
|
||||
|
||||
|
||||
def eval_name_only(
|
||||
name: str,
|
||||
module_name: str,
|
||||
*,
|
||||
locals_: Optional[Mapping[str, Any]] = None,
|
||||
) -> Any:
|
||||
if "." in name:
|
||||
return eval_expression(name, module_name, locals_=locals_)
|
||||
|
||||
try:
|
||||
base_globals: Dict[str, Any] = sys.modules[module_name].__dict__
|
||||
except KeyError as ke:
|
||||
raise NameError(
|
||||
f"Module {module_name} isn't present in sys.modules; can't "
|
||||
f"resolve name {name}"
|
||||
) from ke
|
||||
|
||||
# name only, just look in globals. eval() works perfectly fine here,
|
||||
# however we are seeking to have this be faster, as this occurs for
|
||||
# every Mapper[] keyword, etc. depending on configuration
|
||||
try:
|
||||
return base_globals[name]
|
||||
except KeyError as ke:
|
||||
# check in builtins as well to handle `list`, `set` or `dict`, etc.
|
||||
try:
|
||||
return builtins.__dict__[name]
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
raise NameError(
|
||||
f"Could not locate name {name} in module {module_name}"
|
||||
) from ke
|
||||
|
||||
|
||||
def resolve_name_to_real_class_name(name: str, module_name: str) -> str:
|
||||
try:
|
||||
obj = eval_name_only(name, module_name)
|
||||
except NameError:
|
||||
return name
|
||||
else:
|
||||
return getattr(obj, "__name__", name)
|
||||
|
||||
|
||||
def is_pep593(type_: Optional[Any]) -> bool:
|
||||
return type_ is not None and get_origin(type_) in _type_tuples.Annotated
|
||||
|
||||
|
||||
def is_non_string_iterable(obj: Any) -> TypeGuard[Iterable[Any]]:
|
||||
return isinstance(obj, collections_abc.Iterable) and not isinstance(
|
||||
obj, (str, bytes)
|
||||
)
|
||||
|
||||
|
||||
def is_literal(type_: Any) -> bool:
|
||||
return get_origin(type_) in _type_tuples.Literal
|
||||
|
||||
|
||||
def is_newtype(type_: Optional[_AnnotationScanType]) -> TypeGuard[NewType]:
|
||||
return hasattr(type_, "__supertype__")
|
||||
|
||||
# doesn't work in 3.8, 3.7 as it passes a closure, not an
|
||||
# object instance
|
||||
# isinstance(type, type_instances.NewType)
|
||||
|
||||
|
||||
def is_generic(type_: _AnnotationScanType) -> TypeGuard[GenericProtocol[Any]]:
|
||||
return hasattr(type_, "__args__") and hasattr(type_, "__origin__")
|
||||
|
||||
|
||||
def is_pep695(type_: _AnnotationScanType) -> TypeGuard[TypeAliasType]:
|
||||
# NOTE: a generic TAT does not instance check as TypeAliasType outside of
|
||||
# python 3.10. For sqlalchemy use cases it's fine to consider it a TAT
|
||||
# though.
|
||||
# NOTE: things seems to work also without this additional check
|
||||
if is_generic(type_):
|
||||
return is_pep695(type_.__origin__)
|
||||
return isinstance(type_, _type_instances.TypeAliasType)
|
||||
|
||||
|
||||
def flatten_newtype(type_: NewType) -> Type[Any]:
|
||||
super_type = type_.__supertype__
|
||||
while is_newtype(super_type):
|
||||
super_type = super_type.__supertype__
|
||||
return super_type # type: ignore[return-value]
|
||||
|
||||
|
||||
def pep695_values(type_: _AnnotationScanType) -> Set[Any]:
|
||||
"""Extracts the value from a TypeAliasType, recursively exploring unions
|
||||
and inner TypeAliasType to flatten them into a single set.
|
||||
|
||||
Forward references are not evaluated, so no recursive exploration happens
|
||||
into them.
|
||||
"""
|
||||
_seen = set()
|
||||
|
||||
def recursive_value(inner_type):
|
||||
if inner_type in _seen:
|
||||
# recursion are not supported (at least it's flagged as
|
||||
# an error by pyright). Just avoid infinite loop
|
||||
return inner_type
|
||||
_seen.add(inner_type)
|
||||
if not is_pep695(inner_type):
|
||||
return inner_type
|
||||
value = inner_type.__value__
|
||||
if not is_union(value):
|
||||
return value
|
||||
return [recursive_value(t) for t in value.__args__]
|
||||
|
||||
res = recursive_value(type_)
|
||||
if isinstance(res, list):
|
||||
types = set()
|
||||
stack = deque(res)
|
||||
while stack:
|
||||
t = stack.popleft()
|
||||
if isinstance(t, list):
|
||||
stack.extend(t)
|
||||
else:
|
||||
types.add(None if t is NoneType or is_fwd_none(t) else t)
|
||||
return types
|
||||
else:
|
||||
return {res}
|
||||
|
||||
|
||||
def is_fwd_ref(
|
||||
type_: _AnnotationScanType,
|
||||
check_generic: bool = False,
|
||||
check_for_plain_string: bool = False,
|
||||
) -> TypeGuard[ForwardRef]:
|
||||
if check_for_plain_string and isinstance(type_, str):
|
||||
return True
|
||||
elif isinstance(type_, _type_instances.ForwardRef):
|
||||
return True
|
||||
elif check_generic and is_generic(type_):
|
||||
return any(
|
||||
is_fwd_ref(
|
||||
arg, True, check_for_plain_string=check_for_plain_string
|
||||
)
|
||||
for arg in type_.__args__
|
||||
)
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
@overload
|
||||
def de_optionalize_union_types(type_: str) -> str: ...
|
||||
|
||||
|
||||
@overload
|
||||
def de_optionalize_union_types(type_: Type[Any]) -> Type[Any]: ...
|
||||
|
||||
|
||||
@overload
|
||||
def de_optionalize_union_types(
|
||||
type_: _AnnotationScanType,
|
||||
) -> _AnnotationScanType: ...
|
||||
|
||||
|
||||
def de_optionalize_union_types(
|
||||
type_: _AnnotationScanType,
|
||||
) -> _AnnotationScanType:
|
||||
"""Given a type, filter out ``Union`` types that include ``NoneType``
|
||||
to not include the ``NoneType``.
|
||||
|
||||
Contains extra logic to work on non-flattened unions, unions that contain
|
||||
``None`` (seen in py38, 37)
|
||||
|
||||
"""
|
||||
|
||||
if is_fwd_ref(type_):
|
||||
return _de_optionalize_fwd_ref_union_types(type_, False)
|
||||
|
||||
elif is_union(type_) and includes_none(type_):
|
||||
if compat.py39:
|
||||
typ = set(type_.__args__)
|
||||
else:
|
||||
# py38, 37 - unions are not automatically flattened, can contain
|
||||
# None rather than NoneType
|
||||
stack_of_unions = deque([type_])
|
||||
typ = set()
|
||||
while stack_of_unions:
|
||||
u_typ = stack_of_unions.popleft()
|
||||
for elem in u_typ.__args__:
|
||||
if is_union(elem):
|
||||
stack_of_unions.append(elem)
|
||||
else:
|
||||
typ.add(elem)
|
||||
|
||||
typ.discard(None) # type: ignore[arg-type,unused-ignore]
|
||||
|
||||
typ = {t for t in typ if t is not NoneType and not is_fwd_none(t)}
|
||||
|
||||
return make_union_type(*typ)
|
||||
|
||||
else:
|
||||
return type_
|
||||
|
||||
|
||||
@overload
|
||||
def _de_optionalize_fwd_ref_union_types(
|
||||
type_: ForwardRef, return_has_none: Literal[True]
|
||||
) -> bool: ...
|
||||
|
||||
|
||||
@overload
|
||||
def _de_optionalize_fwd_ref_union_types(
|
||||
type_: ForwardRef, return_has_none: Literal[False]
|
||||
) -> _AnnotationScanType: ...
|
||||
|
||||
|
||||
def _de_optionalize_fwd_ref_union_types(
|
||||
type_: ForwardRef, return_has_none: bool
|
||||
) -> Union[_AnnotationScanType, bool]:
|
||||
"""return the non-optional type for Optional[], Union[None, ...], x|None,
|
||||
etc. without de-stringifying forward refs.
|
||||
|
||||
unfortunately this seems to require lots of hardcoded heuristics
|
||||
|
||||
"""
|
||||
|
||||
annotation = type_.__forward_arg__
|
||||
|
||||
mm = re.match(r"^(.+?)\[(.+)\]$", annotation)
|
||||
if mm:
|
||||
g1 = mm.group(1).split(".")[-1]
|
||||
if g1 == "Optional":
|
||||
return True if return_has_none else ForwardRef(mm.group(2))
|
||||
elif g1 == "Union":
|
||||
if "[" in mm.group(2):
|
||||
# cases like "Union[Dict[str, int], int, None]"
|
||||
elements: list[str] = []
|
||||
current: list[str] = []
|
||||
ignore_comma = 0
|
||||
for char in mm.group(2):
|
||||
if char == "[":
|
||||
ignore_comma += 1
|
||||
elif char == "]":
|
||||
ignore_comma -= 1
|
||||
elif ignore_comma == 0 and char == ",":
|
||||
elements.append("".join(current).strip())
|
||||
current.clear()
|
||||
continue
|
||||
current.append(char)
|
||||
else:
|
||||
elements = re.split(r",\s*", mm.group(2))
|
||||
parts = [ForwardRef(elem) for elem in elements if elem != "None"]
|
||||
if return_has_none:
|
||||
return len(elements) != len(parts)
|
||||
else:
|
||||
return make_union_type(*parts) if parts else Never # type: ignore[return-value] # noqa: E501
|
||||
else:
|
||||
return False if return_has_none else type_
|
||||
|
||||
pipe_tokens = re.split(r"\s*\|\s*", annotation)
|
||||
has_none = "None" in pipe_tokens
|
||||
if return_has_none:
|
||||
return has_none
|
||||
if has_none:
|
||||
anno_str = "|".join(p for p in pipe_tokens if p != "None")
|
||||
return ForwardRef(anno_str) if anno_str else Never # type: ignore[return-value] # noqa: E501
|
||||
|
||||
return type_
|
||||
|
||||
|
||||
def make_union_type(*types: _AnnotationScanType) -> Type[Any]:
|
||||
"""Make a Union type."""
|
||||
|
||||
return Union[types] # type: ignore
|
||||
|
||||
|
||||
def includes_none(type_: Any) -> bool:
|
||||
"""Returns if the type annotation ``type_`` allows ``None``.
|
||||
|
||||
This function supports:
|
||||
* forward refs
|
||||
* unions
|
||||
* pep593 - Annotated
|
||||
* pep695 - TypeAliasType (does not support looking into
|
||||
fw reference of other pep695)
|
||||
* NewType
|
||||
* plain types like ``int``, ``None``, etc
|
||||
"""
|
||||
if is_fwd_ref(type_):
|
||||
return _de_optionalize_fwd_ref_union_types(type_, True)
|
||||
if is_union(type_):
|
||||
return any(includes_none(t) for t in get_args(type_))
|
||||
if is_pep593(type_):
|
||||
return includes_none(get_args(type_)[0])
|
||||
if is_pep695(type_):
|
||||
return any(includes_none(t) for t in pep695_values(type_))
|
||||
if is_newtype(type_):
|
||||
return includes_none(type_.__supertype__)
|
||||
try:
|
||||
return type_ in (NoneType, None) or is_fwd_none(type_)
|
||||
except TypeError:
|
||||
# if type_ is Column, mapped_column(), etc. the use of "in"
|
||||
# resolves to ``__eq__()`` which then gives us an expression object
|
||||
# that can't resolve to boolean. just catch it all via exception
|
||||
return False
|
||||
|
||||
|
||||
def is_a_type(type_: Any) -> bool:
|
||||
return (
|
||||
isinstance(type_, type)
|
||||
or get_origin(type_) is not None
|
||||
or getattr(type_, "__module__", None)
|
||||
in ("typing", "typing_extensions")
|
||||
or type(type_).__mro__[0].__module__ in ("typing", "typing_extensions")
|
||||
)
|
||||
|
||||
|
||||
def is_union(type_: Any) -> TypeGuard[ArgsTypeProtocol]:
|
||||
return is_origin_of(type_, "Union", "UnionType")
|
||||
|
||||
|
||||
def is_origin_of_cls(
|
||||
type_: Any, class_obj: Union[Tuple[Type[Any], ...], Type[Any]]
|
||||
) -> bool:
|
||||
"""return True if the given type has an __origin__ that shares a base
|
||||
with the given class"""
|
||||
|
||||
origin = get_origin(type_)
|
||||
if origin is None:
|
||||
return False
|
||||
|
||||
return isinstance(origin, type) and issubclass(origin, class_obj)
|
||||
|
||||
|
||||
def is_origin_of(
|
||||
type_: Any, *names: str, module: Optional[str] = None
|
||||
) -> bool:
|
||||
"""return True if the given type has an __origin__ with the given name
|
||||
and optional module."""
|
||||
|
||||
origin = get_origin(type_)
|
||||
if origin is None:
|
||||
return False
|
||||
|
||||
return _get_type_name(origin) in names and (
|
||||
module is None or origin.__module__.startswith(module)
|
||||
)
|
||||
|
||||
|
||||
def _get_type_name(type_: Type[Any]) -> str:
|
||||
if compat.py310:
|
||||
return type_.__name__
|
||||
else:
|
||||
typ_name = getattr(type_, "__name__", None)
|
||||
if typ_name is None:
|
||||
typ_name = getattr(type_, "_name", None)
|
||||
|
||||
return typ_name # type: ignore
|
||||
|
||||
|
||||
class DescriptorProto(Protocol):
|
||||
def __get__(self, instance: object, owner: Any) -> Any: ...
|
||||
|
||||
def __set__(self, instance: Any, value: Any) -> None: ...
|
||||
|
||||
def __delete__(self, instance: Any) -> None: ...
|
||||
|
||||
|
||||
_DESC = TypeVar("_DESC", bound=DescriptorProto)
|
||||
|
||||
|
||||
class DescriptorReference(Generic[_DESC]):
|
||||
"""a descriptor that refers to a descriptor.
|
||||
|
||||
used for cases where we need to have an instance variable referring to an
|
||||
object that is itself a descriptor, which typically confuses typing tools
|
||||
as they don't know when they should use ``__get__`` or not when referring
|
||||
to the descriptor assignment as an instance variable. See
|
||||
sqlalchemy.orm.interfaces.PropComparator.prop
|
||||
|
||||
"""
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
||||
def __get__(self, instance: object, owner: Any) -> _DESC: ...
|
||||
|
||||
def __set__(self, instance: Any, value: _DESC) -> None: ...
|
||||
|
||||
def __delete__(self, instance: Any) -> None: ...
|
||||
|
||||
|
||||
_DESC_co = TypeVar("_DESC_co", bound=DescriptorProto, covariant=True)
|
||||
|
||||
|
||||
class RODescriptorReference(Generic[_DESC_co]):
|
||||
"""a descriptor that refers to a descriptor.
|
||||
|
||||
same as :class:`.DescriptorReference` but is read-only, so that subclasses
|
||||
can define a subtype as the generically contained element
|
||||
|
||||
"""
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
||||
def __get__(self, instance: object, owner: Any) -> _DESC_co: ...
|
||||
|
||||
def __set__(self, instance: Any, value: Any) -> NoReturn: ...
|
||||
|
||||
def __delete__(self, instance: Any) -> NoReturn: ...
|
||||
|
||||
|
||||
_FN = TypeVar("_FN", bound=Optional[Callable[..., Any]])
|
||||
|
||||
|
||||
class CallableReference(Generic[_FN]):
|
||||
"""a descriptor that refers to a callable.
|
||||
|
||||
works around mypy's limitation of not allowing callables assigned
|
||||
as instance variables
|
||||
|
||||
|
||||
"""
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
||||
def __get__(self, instance: object, owner: Any) -> _FN: ...
|
||||
|
||||
def __set__(self, instance: Any, value: _FN) -> None: ...
|
||||
|
||||
def __delete__(self, instance: Any) -> None: ...
|
||||
|
||||
|
||||
class _TypingInstances:
|
||||
def __getattr__(self, key: str) -> tuple[type, ...]:
|
||||
types = tuple(
|
||||
{
|
||||
t
|
||||
for t in [
|
||||
getattr(typing, key, None),
|
||||
getattr(typing_extensions, key, None),
|
||||
]
|
||||
if t is not None
|
||||
}
|
||||
)
|
||||
if not types:
|
||||
raise AttributeError(key)
|
||||
self.__dict__[key] = types
|
||||
return types
|
||||
|
||||
|
||||
_type_tuples = _TypingInstances()
|
||||
if TYPE_CHECKING:
|
||||
_type_instances = typing_extensions
|
||||
else:
|
||||
_type_instances = _type_tuples
|
||||
|
||||
LITERAL_TYPES = _type_tuples.Literal
|
||||
Reference in New Issue
Block a user