summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/util/langhelpers.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2022-04-07 12:37:23 -0400
committermike bayer <mike_mp@zzzcomputing.com>2022-04-12 02:09:50 +0000
commitaa9cd878e8249a4a758c7f968e929e92fede42a5 (patch)
tree1be1c9dc24dd247a150be55d65bfc56ebaf111bc /lib/sqlalchemy/util/langhelpers.py
parent98eae4e181cb2d1bbc67ec834bfad29dcba7f461 (diff)
downloadsqlalchemy-aa9cd878e8249a4a758c7f968e929e92fede42a5.tar.gz
pep-484: session, instancestate, etc
Also adds some fixes to annotation-based mapping that have come up, as well as starts to add more pep-484 test cases Change-Id: Ia722bbbc7967a11b23b66c8084eb61df9d233fee
Diffstat (limited to 'lib/sqlalchemy/util/langhelpers.py')
-rw-r--r--lib/sqlalchemy/util/langhelpers.py173
1 files changed, 85 insertions, 88 deletions
diff --git a/lib/sqlalchemy/util/langhelpers.py b/lib/sqlalchemy/util/langhelpers.py
index 3e89c72bb..2cb9c45d6 100644
--- a/lib/sqlalchemy/util/langhelpers.py
+++ b/lib/sqlalchemy/util/langhelpers.py
@@ -12,6 +12,7 @@ modules, classes, hierarchies, attributes, functions, and methods.
from __future__ import annotations
import collections
+import enum
from functools import update_wrapper
import hashlib
import inspect
@@ -671,13 +672,13 @@ def format_argspec_init(method, grouped=True):
def create_proxy_methods(
- target_cls,
- target_cls_sphinx_name,
- proxy_cls_sphinx_name,
- classmethods=(),
- methods=(),
- attributes=(),
-):
+ target_cls: Type[Any],
+ target_cls_sphinx_name: str,
+ proxy_cls_sphinx_name: str,
+ classmethods: Sequence[str] = (),
+ methods: Sequence[str] = (),
+ attributes: Sequence[str] = (),
+) -> Callable[[_T], _T]:
"""A class decorator indicating attributes should refer to a proxy
class.
@@ -1539,24 +1540,50 @@ class hybridmethod(Generic[_T]):
return self
-class _symbol(int):
+class symbol(int):
+ """A constant symbol.
+
+ >>> symbol('foo') is symbol('foo')
+ True
+ >>> symbol('foo')
+ <symbol 'foo>
+
+ A slight refinement of the MAGICCOOKIE=object() pattern. The primary
+ advantage of symbol() is its repr(). They are also singletons.
+
+ Repeated calls of symbol('name') will all return the same instance.
+
+ In SQLAlchemy 2.0, symbol() is used for the implementation of
+ ``_FastIntFlag``, but otherwise should be mostly replaced by
+ ``enum.Enum`` and variants.
+
+
+ """
+
name: str
+ symbols: Dict[str, symbol] = {}
+ _lock = threading.Lock()
+
def __new__(
cls,
name: str,
doc: Optional[str] = None,
canonical: Optional[int] = None,
- ) -> "_symbol":
- """Construct a new named symbol."""
- assert isinstance(name, str)
- if canonical is None:
- canonical = hash(name)
- v = int.__new__(_symbol, canonical)
- v.name = name
- if doc:
- v.__doc__ = doc
- return v
+ ) -> symbol:
+ with cls._lock:
+ sym = cls.symbols.get(name)
+ if sym is None:
+ assert isinstance(name, str)
+ if canonical is None:
+ canonical = hash(name)
+ sym = int.__new__(symbol, canonical)
+ sym.name = name
+ if doc:
+ sym.__doc__ = doc
+
+ cls.symbols[name] = sym
+ return sym
def __reduce__(self):
return symbol, (self.name, "x", int(self))
@@ -1565,90 +1592,60 @@ class _symbol(int):
return repr(self)
def __repr__(self):
- return "symbol(%r)" % self.name
+ return f"symbol({self.name!r})"
-_symbol.__name__ = "symbol"
+class _IntFlagMeta(type):
+ def __init__(
+ cls,
+ classname: str,
+ bases: Tuple[Type[Any], ...],
+ dict_: Dict[str, Any],
+ **kw: Any,
+ ) -> None:
+ items: List[symbol]
+ cls._items = items = []
+ for k, v in dict_.items():
+ if isinstance(v, int):
+ sym = symbol(k, canonical=v)
+ elif not k.startswith("_"):
+ raise TypeError("Expected integer values for IntFlag")
+ else:
+ continue
+ setattr(cls, k, sym)
+ items.append(sym)
+ def __iter__(self) -> Iterator[symbol]:
+ return iter(self._items)
-class symbol:
- """A constant symbol.
- >>> symbol('foo') is symbol('foo')
- True
- >>> symbol('foo')
- <symbol 'foo>
+class _FastIntFlag(metaclass=_IntFlagMeta):
+ """An 'IntFlag' copycat that isn't slow when performing bitwise
+ operations.
- A slight refinement of the MAGICCOOKIE=object() pattern. The primary
- advantage of symbol() is its repr(). They are also singletons.
+ the ``FastIntFlag`` class will return ``enum.IntFlag`` under TYPE_CHECKING
+ and ``_FastIntFlag`` otherwise.
- Repeated calls of symbol('name') will all return the same instance.
+ """
- The optional ``doc`` argument assigns to ``__doc__``. This
- is strictly so that Sphinx autoattr picks up the docstring we want
- (it doesn't appear to pick up the in-module docstring if the datamember
- is in a different module - autoattribute also blows up completely).
- If Sphinx fixes/improves this then we would no longer need
- ``doc`` here.
- """
+if TYPE_CHECKING:
+ from enum import IntFlag
- symbols: Dict[str, "_symbol"] = {}
- _lock = threading.Lock()
+ FastIntFlag = IntFlag
+else:
+ FastIntFlag = _FastIntFlag
- def __new__( # type: ignore[misc]
- cls,
- name: str,
- doc: Optional[str] = None,
- canonical: Optional[int] = None,
- ) -> _symbol:
- with cls._lock:
- sym = cls.symbols.get(name)
- if sym is None:
- cls.symbols[name] = sym = _symbol(name, doc, canonical)
- return sym
- @classmethod
- def parse_user_argument(
- cls, arg, choices, name, resolve_symbol_names=False
- ):
- """Given a user parameter, parse the parameter into a chosen symbol.
-
- The user argument can be a string name that matches the name of a
- symbol, or the symbol object itself, or any number of alternate choices
- such as True/False/ None etc.
-
- :param arg: the user argument.
- :param choices: dictionary of symbol object to list of possible
- entries.
- :param name: name of the argument. Used in an :class:`.ArgumentError`
- that is raised if the parameter doesn't match any available argument.
- :param resolve_symbol_names: include the name of each symbol as a valid
- entry.
-
- """
- # note using hash lookup is tricky here because symbol's `__hash__`
- # is its int value which we don't want included in the lookup
- # explicitly, so we iterate and compare each.
- for sym, choice in choices.items():
- if arg is sym:
- return sym
- elif resolve_symbol_names and arg == sym.name:
- return sym
- elif arg in choice:
- return sym
-
- if arg is None:
- return None
-
- raise exc.ArgumentError("Invalid value for '%s': %r" % (name, arg))
+_E = TypeVar("_E", bound=enum.Enum)
def parse_user_argument_for_enum(
arg: Any,
- choices: Dict[_T, List[Any]],
+ choices: Dict[_E, List[Any]],
name: str,
-) -> Optional[_T]:
+ resolve_symbol_names: bool = False,
+) -> Optional[_E]:
"""Given a user parameter, parse the parameter into a chosen value
from a list of choice objects, typically Enum values.
@@ -1663,18 +1660,18 @@ def parse_user_argument_for_enum(
that is raised if the parameter doesn't match any available argument.
"""
- # TODO: use whatever built in thing Enum provides for this,
- # if applicable
for enum_value, choice in choices.items():
if arg is enum_value:
return enum_value
+ elif resolve_symbol_names and arg == enum_value.name:
+ return enum_value
elif arg in choice:
return enum_value
if arg is None:
return None
- raise exc.ArgumentError("Invalid value for '%s': %r" % (name, arg))
+ raise exc.ArgumentError(f"Invalid value for '{name}': {arg!r}")
_creation_order = 1