self.update_execution_options(**execution_options)
def _lru_size_alert(self, cache: util.LRUCache[Any, Any]) -> None:
- if self._should_log_info:
+ if self._should_log_info():
self.logger.info(
"Compiled cache size pruning from %d items to %d. "
"Increase cache size to reduce the frequency of pruning.",
This is an internal dialect method. Applications should use
:meth:`.Inspector.get_columns`.
+
"""
+ raise NotImplementedError()
+
def get_multi_columns(
self,
connection: Connection,
...
-class _SetterProtocol(Protocol[_T_co]):
+# mypy 0.990 we are no longer allowed to make this Protocol[_T_con]
+class _SetterProtocol(Protocol):
...
-class _PlainSetterProtocol(_SetterProtocol[_T_con]):
+class _PlainSetterProtocol(_SetterProtocol, Protocol[_T_con]):
def __call__(self, instance: Any, value: _T_con) -> None:
...
-class _DictSetterProtocol(_SetterProtocol[_T_con]):
+class _DictSetterProtocol(_SetterProtocol, Protocol[_T_con]):
def __call__(self, instance: Any, key: Any, value: _T_con) -> None:
...
-class _CreatorProtocol(Protocol[_T_co]):
+# mypy 0.990 we are no longer allowed to make this Protocol[_T_con]
+class _CreatorProtocol(Protocol):
...
-class _PlainCreatorProtocol(_CreatorProtocol[_T_con]):
+class _PlainCreatorProtocol(_CreatorProtocol, Protocol[_T_con]):
def __call__(self, value: _T_con) -> Any:
...
-class _KeyCreatorProtocol(_CreatorProtocol[_T_con]):
+class _KeyCreatorProtocol(_CreatorProtocol, Protocol[_T_con]):
def __call__(self, key: Any, value: Optional[_T_con]) -> Any:
...
self,
collection_class: Optional[Type[Any]],
assoc_instance: AssociationProxyInstance[Any],
- ) -> Tuple[_GetterProtocol[Any], _SetterProtocol[Any]]:
+ ) -> Tuple[_GetterProtocol[Any], _SetterProtocol]:
...
def __call__(
self,
lazy_collection: _LazyCollectionProtocol[Any],
- creator: _CreatorProtocol[Any],
+ creator: _CreatorProtocol,
value_attr: str,
parent: AssociationProxyInstance[Any],
) -> Any:
"""describes the interface of :class:`.AssociationProxy`
without including descriptor methods in the interface."""
- creator: Optional[_CreatorProtocol[Any]]
+ creator: Optional[_CreatorProtocol]
key: str
target_collection: str
value_attr: str
def _default_getset(
self, collection_class: Any
- ) -> Tuple[_GetterProtocol[Any], _SetterProtocol[Any]]:
+ ) -> Tuple[_GetterProtocol[Any], _SetterProtocol]:
...
self,
target_collection: str,
attr: str,
- creator: Optional[_CreatorProtocol[Any]] = None,
+ creator: Optional[_CreatorProtocol] = None,
getset_factory: Optional[_GetSetFactoryProtocol] = None,
proxy_factory: Optional[_ProxyFactoryProtocol] = None,
proxy_bulk_set: Optional[_ProxyBulkSetProtocol] = None,
def _default_getset(
self, collection_class: Any
- ) -> Tuple[_GetterProtocol[Any], _SetterProtocol[Any]]:
+ ) -> Tuple[_GetterProtocol[Any], _SetterProtocol]:
attr = self.value_attr
_getter = operator.attrgetter(attr)
def _default_getset(
self, collection_class: Any
- ) -> Tuple[_GetterProtocol[Any], _SetterProtocol[Any]]:
+ ) -> Tuple[_GetterProtocol[Any], _SetterProtocol]:
attr = self.value_attr
_getter = operator.attrgetter(attr)
creator = (
self.parent.creator
if self.parent.creator is not None
- else cast("_CreatorProtocol[_T]", self.target_class)
+ else cast("_CreatorProtocol", self.target_class)
)
collection_class = util.duck_type_collection(lazy_collection())
creator = (
self.parent.creator
and self.parent.creator
- or cast(_CreatorProtocol[Any], self.target_class)
+ or cast(_CreatorProtocol, self.target_class)
)
if self.parent.getset_factory:
getter: _GetterProtocol[_IT]
"""A function. Given an associated object, return the 'value'."""
- creator: _CreatorProtocol[_IT]
+ creator: _CreatorProtocol
"""
A function that creates new target entities. Given one parameter:
value. This assertion is assumed::
"""
parent: AssociationProxyInstance[_IT]
- setter: _SetterProtocol[_IT]
+ setter: _SetterProtocol
"""A function. Given an associated object and a value, store that
value on the object.
"""
def __init__(
self,
lazy_collection: _LazyCollectionProtocol[_IT],
- creator: _CreatorProtocol[_IT],
+ creator: _CreatorProtocol,
getter: _GetterProtocol[_IT],
- setter: _SetterProtocol[_IT],
+ setter: _SetterProtocol,
parent: AssociationProxyInstance[_IT],
):
"""Constructs an _AssociationCollection.
if crit:
eval_condition = evaluator_compiler.process(*crit)
else:
-
- def eval_condition(obj):
+ # workaround for mypy https://github.com/python/mypy/issues/14027
+ def _eval_condition(obj):
return True
+ eval_condition = _eval_condition
+
return eval_condition
@classmethod
-# ext/declarative/api.py
+# orm/declarative/api.py
# Copyright (C) 2005-2022 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
if isinstance(cls, type):
class_dict["__doc__"] = cls.__doc__
- if self.constructor:
+ if self.constructor is not None:
class_dict["__init__"] = self.constructor
class_dict["__abstract__"] = True
lambda ref: _finalize_fairy(
None, rec, pool, ref, echo, transaction_was_reset=False
)
- if _finalize_fairy
+ if _finalize_fairy is not None
else None,
)
_strong_ref_connection_records[ref] = rec
# at all if the key were present in the parameters
if autoinc_key in self.binds:
- def autoinc_getter(lastrowid, parameters):
+ def _autoinc_getter(lastrowid, parameters):
param_value = parameters.get(autoinc_key, lastrowid)
if param_value is not None:
# they supplied non-None parameter, use that.
# use lastrowid
return lastrowid
+ # work around mypy https://github.com/python/mypy/issues/14027
+ autoinc_getter = _autoinc_getter
+
else:
lastrowid_processor = None
return row_fn(
(
autoinc_getter(lastrowid, parameters)
- if autoinc_getter
+ if autoinc_getter is not None
else lastrowid
)
if col is autoinc_col
assert right_attrname is not None
dispatch = self.dispatch(left_visit_sym)
- assert dispatch, (
+ assert dispatch is not None, (
f"{self.__class__} has no dispatch for "
f"'{self._dispatch_lookup[left_visit_sym]}'"
)
names = []
for attrname, visit_sym in internal_dispatch:
meth = self.dispatch(visit_sym)
- if meth:
+ if meth is not None:
visit_name = _dispatch_lookup[visit_sym]
names.append((attrname, visit_name))
else:
return specimen.__emulates__ # type: ignore
- isa = isinstance(specimen, type) and issubclass or isinstance
+ isa = issubclass if isinstance(specimen, type) else isinstance
if isa(specimen, list):
return list
elif isa(specimen, set):
+# util/typing.py
+# Copyright (C) 2022 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
from typing import overload
from typing import Tuple
from typing import Type
+from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union
"""
- def __get__(self, instance: object, owner: Any) -> _DESC:
- ...
+ if TYPE_CHECKING:
- def __set__(self, instance: Any, value: _DESC) -> None:
- ...
+ def __get__(self, instance: object, owner: Any) -> _DESC:
+ ...
- def __delete__(self, instance: Any) -> None:
- ...
+ def __set__(self, instance: Any, value: _DESC) -> None:
+ ...
+
+ def __delete__(self, instance: Any) -> None:
+ ...
_DESC_co = TypeVar("_DESC_co", bound=DescriptorProto, covariant=True)
"""
- def __get__(self, instance: object, owner: Any) -> _DESC_co:
- ...
+ if TYPE_CHECKING:
- def __set__(self, instance: Any, value: Any) -> NoReturn:
- ...
+ def __get__(self, instance: object, owner: Any) -> _DESC_co:
+ ...
- def __delete__(self, instance: Any) -> NoReturn:
- ...
+ def __set__(self, instance: Any, value: Any) -> NoReturn:
+ ...
+
+ def __delete__(self, instance: Any) -> NoReturn:
+ ...
_FN = TypeVar("_FN", bound=Optional[Callable[..., Any]])
"""
- def __get__(self, instance: object, owner: Any) -> _FN:
- ...
+ if TYPE_CHECKING:
- def __set__(self, instance: Any, value: _FN) -> None:
- ...
+ def __get__(self, instance: object, owner: Any) -> _FN:
+ ...
- def __delete__(self, instance: Any) -> None:
- ...
+ def __set__(self, instance: Any, value: _FN) -> None:
+ ...
+
+ def __delete__(self, instance: Any) -> None:
+ ...
# $def ro_descriptor_reference(fn: Callable[])
),
]
- args.append(path)
+ # mypy as of 0.990 is more aggressively blocking messaging
+ # for paths that are in sys.path, and as pytest puts currdir,
+ # test/ etc in sys.path, just copy the source file to the
+ # tempdir we are working in so that we don't have to try to
+ # manipulate sys.path and/or guess what mypy is doing
+ filename = os.path.basename(path)
+ test_program = os.path.join(cachedir, filename)
+ shutil.copyfile(path, test_program)
+ args.append(test_program)
# I set this locally but for the suite here needs to be
# disabled
not_located = []
if expected_messages:
- eq_(result[2], 1, msg=result)
+ # mypy 0.990 changed how return codes work, so don't assume a
+ # 1 or a 0 return code here, could be either depending on if
+ # errors were generated or not
output = []
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
