def close(self) -> None:
if self._cursor is not None:
await_(self._cursor.close())
- self._cursor = None # type: ignore
+ self._cursor = None # type: ignore[assignment]
def fetchone(self) -> Optional[Any]:
return await_(self._cursor.fetchone())
module: Any = __import__(
"sqlalchemy.dialects.mysql.mariadb"
).dialects.mysql.mariadb
- return module.loader(driver) # type: ignore
+ return module.loader(driver) # type: ignore[no-any-return]
else:
module = __import__("sqlalchemy.dialects.%s" % (dialect,)).dialects
module = getattr(module, dialect)
await_(self._connection.autocommit(value))
def get_autocommit(self) -> bool:
- return self._connection.get_autocommit() # type: ignore
+ return self._connection.get_autocommit() # type: ignore[no-any-return]
def close(self) -> None:
await_(self._connection.ensure_closed())
return "not connected" in str_e
def _found_rows_client_flag(self) -> int:
- from pymysql.constants import CLIENT # type: ignore
+ from pymysql.constants import CLIENT # type: ignore[import-untyped]
return CLIENT.FOUND_ROWS # type: ignore[no-any-return]
await_(self._connection.autocommit(value))
def get_autocommit(self) -> bool:
- return self._connection.get_autocommit() # type: ignore
+ return self._connection.get_autocommit() # type: ignore[no-any-return]
def close(self) -> None:
await_(self._connection.ensure_closed())
)
def _found_rows_client_flag(self) -> int:
- from asyncmy.constants import CLIENT # type: ignore
+ from asyncmy.constants import CLIENT # type: ignore[import-not-found]
return CLIENT.FOUND_ROWS # type: ignore[no-any-return]
if isinstance(
e,
(
- self.dbapi.OperationalError, # type: ignore
- self.dbapi.ProgrammingError, # type: ignore
- self.dbapi.InterfaceError, # type: ignore
+ self.dbapi.OperationalError, # type: ignore[union-attr]
+ self.dbapi.ProgrammingError, # type: ignore[union-attr]
+ self.dbapi.InterfaceError, # type: ignore[union-attr]
),
) and self._extract_error_code(e) in (
1927,
):
return True
elif isinstance(
- e, (self.dbapi.InterfaceError, self.dbapi.InternalError) # type: ignore # noqa: E501
+ e, (self.dbapi.InterfaceError, self.dbapi.InternalError) # type: ignore[union-attr] # noqa: E501
):
# if underlying connection is closed,
# this is the error you get
#
# there's more "doesn't exist" kinds of messages but they are
# less clear if mysql 8 would suddenly start using one of those
- if self._extract_error_code(e.orig) in (1146, 1049, 1051): # type: ignore # noqa: E501
+ if self._extract_error_code(e.orig) in (1146, 1049, 1051): # type: ignore[arg-type, operator] # noqa: E501
return False
raise
if schema is not None:
current_schema: str = schema
else:
- current_schema = self.default_schema_name # type: ignore
+ current_schema = self.default_schema_name # type: ignore[assignment] # noqa: E501
charset = self._connection_charset
def lower(s: str) -> str:
return s
- default_schema_name: str = connection.dialect.default_schema_name # type: ignore # noqa: E501
+ default_schema_name: str = connection.dialect.default_schema_name # type: ignore[assignment] # noqa: E501
# NOTE: using (table_schema, table_name, lower(column_name)) in (...)
# is very slow since mysql does not seem able to properly use indexse.
if value is not None and not isinstance(value, (int, str)):
value = ",".join(value)
if super_convert:
- return super_convert(value) # type: ignore
+ return super_convert(value) # type: ignore[arg-type, no-any-return] # noqa: E501
else:
return value
# supports_sane_rowcount.
if self.dbapi is not None:
try:
- from mysql.connector import constants # type: ignore
+ from mysql.connector import constants # type: ignore[import-not-found] # noqa: E501
ClientFlag = constants.ClientFlag
return None
def _detect_charset(self, connection: Connection) -> str:
- return connection.connection.charset # type: ignore
+ return connection.connection.charset # type: ignore[no-any-return]
def _extract_error_code(self, exception: BaseException) -> int:
- return exception.errno # type: ignore
+ return exception.errno # type: ignore[attr-defined, no-any-return]
def is_disconnect(
self,
except (AttributeError, ImportError):
return None
else:
- return CLIENT_FLAGS.FOUND_ROWS # type: ignore
+ return CLIENT_FLAGS.FOUND_ROWS # type: ignore[no-any-return]
else:
return None
)
def _setup_getitem(self, index: Any) -> Any:
- return GETITEM, index, self.type.text_type # type: ignore
+ return GETITEM, index, self.type.text_type # type: ignore[attr-defined] # noqa: E501
def defined(self, key: Any) -> Any:
"""Boolean expression. Test for presence of a non-NULL value for
)
if self.upper is None:
- return ( # type: ignore
+ return ( # type: ignore[no-any-return]
value > self.lower
if self.bounds[0] == "("
else value >= self.lower
)
- return ( # type: ignore
+ return ( # type: ignore[no-any-return]
value > self.lower
if self.bounds[0] == "("
else value >= self.lower
return False
if bound1 == "]":
if bound2 == "[":
- return value1 == value2 - step # type: ignore
+ return value1 == value2 - step # type: ignore[no-any-return] # noqa: E501
else:
return value1 == value2
else:
if bound2 == "[":
return value1 == value2
else:
- return value1 == value2 - step # type: ignore
+ return value1 == value2 - step # type: ignore[no-any-return] # noqa: E501
elif res == 0:
# Cover cases like [0,0] -|- [1,] and [0,2) -|- (1,3]
if (
return "empty"
l, r = self.lower, self.upper
- l = "" if l is None else l # type: ignore
- r = "" if r is None else r # type: ignore
+ l = "" if l is None else l # type: ignore[assignment]
+ r = "" if r is None else r # type: ignore[assignment]
b0, b1 = cast("Tuple[str, str]", self.bounds)
# The adapt() operation here is cached per type-class-per-dialect,
# so is not much of a performance concern
visit_name = self.__visit_name__
- return type( # type: ignore
+ return type( # type: ignore[no-any-return]
f"{visit_name}RangeImpl",
(cls, self.__class__),
{"__visit_name__": visit_name},
return pool.SingletonThreadPool
def _get_server_version_info(self, connection: Any) -> VersionInfoType:
- return self.dbapi.sqlite_version_info # type: ignore
+ return self.dbapi.sqlite_version_info # type: ignore[no-any-return, union-attr] # noqa: E501
_isolation_lookup = SQLiteDialect._isolation_lookup.union(
{
made_rows, [], uniques, strategy
)
else:
- interim_rows = made_rows # type: ignore
+ interim_rows = made_rows # type: ignore[assignment]
if post_creational_filter is not None:
interim_rows = [
uniques.add(hashed)
if post_creational_filter is not None:
obj = post_creational_filter(obj)
- return obj # type: ignore
+ return obj # type: ignore[return-value]
else:
)
if post_creational_filter is not None:
interim_row = post_creational_filter(interim_row)
- return interim_row # type: ignore
+ return interim_row # type: ignore[return-value]
return onerow
row = post_creational_filter(row)
if scalar and make_row is not None:
- return row[0] # type: ignore
+ return row[0] # type: ignore[no-any-return]
else:
- return row # type: ignore
+ return row # type: ignore[return-value]
def _iter_impl(self) -> Iterator[_R]:
return self._iterator_getter()
strat = kwargs.pop("strategy")
if strat == "mock":
# this case is deprecated
- return create_mock_engine(url, **kwargs) # type: ignore
+ return create_mock_engine(url, **kwargs) # type: ignore[return-value] # noqa: E501
else:
raise exc.ArgumentError("unknown strategy: %r" % strat)
return value
else:
- pop_kwarg = kwargs.pop # type: ignore
+ pop_kwarg = kwargs.pop # type: ignore[assignment]
dialect_args = {}
# consume dialect arguments from kwargs
result_columns = tuplefilter(result_columns)
num_ctx_cols = len(result_columns)
else:
- result_columns = cols_are_ordered = ( # type: ignore
+ result_columns = cols_are_ordered = ( # type: ignore[assignment]
num_ctx_cols
) = ad_hoc_textual = loose_column_name_matching = (
textual_ordered
if not self._soft_closed:
cursor = self.cursor
- self.cursor = None # type: ignore
+ self.cursor = None # type: ignore[assignment]
self.connection._safe_close_cursor(cursor)
self._soft_closed = True
use_insertmanyvalues: Optional[bool] = None,
# util.deprecated_params decorator cannot render the
# Linting.NO_LINTING constant
- compiler_linting: Linting = int(compiler.NO_LINTING), # type: ignore
+ compiler_linting: Linting = int(compiler.NO_LINTING), # type: ignore[assignment] # noqa: E501
server_side_cursors: bool = False,
skip_autocommit_rollback: bool = False,
**kwargs: Any,
self.is_text = compiled.isplaintext
if ii or iu or id_:
- dml_statement = compiled.compile_state.statement # type: ignore
+ dml_statement = compiled.compile_state.statement # type: ignore[union-attr] # noqa: E501
if TYPE_CHECKING:
assert isinstance(dml_statement, UpdateBase)
self.is_crud = True
self._expanded_parameters = expanded_state.parameter_expansion
- flattened_processors = dict(processors) # type: ignore
+ flattened_processors = dict(processors) # type: ignore[arg-type]
flattened_processors.update(expanded_state.processors)
positiontup = expanded_state.positiontup
elif compiled.positional:
parameters = self.dialect.execute_sequence_format(
[
(
- processors[key](compiled_params[key]) # type: ignore
+ processors[key](compiled_params[key]) # type: ignore[operator] # noqa: E501
if key in processors
else compiled_params[key]
)
else:
parameters = {
key: (
- processors[key](compiled_params[key]) # type: ignore
+ processors[key](compiled_params[key]) # type: ignore[assignment, operator] # noqa: E501
if key in processors
else compiled_params[key]
)
if identifier == "before_execute":
orig_fn = fn
- def wrap_before_execute( # type: ignore
+ def wrap_before_execute( # type: ignore[no-untyped-def]
conn, clauseelement, multiparams, params, execution_options
):
orig_fn(
elif identifier == "before_cursor_execute":
orig_fn = fn
- def wrap_before_cursor_execute( # type: ignore
+ def wrap_before_cursor_execute( # type: ignore[no-untyped-def]
conn, cursor, statement, parameters, context, executemany
):
orig_fn(
_tuplefilter = tuplegetter(*_translated_indexes)
else:
_translated_indexes = _tuplefilter = None
- self.__init__( # type: ignore
+ self.__init__( # type: ignore[misc]
state["_keys"],
_translated_indexes=_translated_indexes,
_tuplefilter=_tuplefilter,
workaround for SQLAlchemy 2.1.
"""
- return self # type: ignore
+ return self # type: ignore[return-value]
@deprecated(
"2.1.0",
"""
- return self # type: ignore
+ return self # type: ignore[return-value]
def _raw_row_iterator(self) -> Iterator[_RowData]:
"""Return a safe iterator that yields raw row data.
def _soft_close(self, hard: bool = False, **kw: Any) -> None:
super()._soft_close(hard=hard, **kw)
- self.chunks = lambda size: [] # type: ignore
+ self.chunks = lambda size: [] # type: ignore[assignment, return-value]
def _fetchmany_impl(
self, size: Optional[int] = None
if components["port"]:
components["port"] = int(components["port"])
- return URL.create(name, **components) # type: ignore
+ return URL.create(name, **components) # type: ignore[arg-type]
else:
raise exc.ArgumentError(
"""
@util.decorator
- def decorated(fn, self, connection): # type: ignore
+ def decorated(fn, self, connection): # type: ignore[no-untyped-def]
connection = connection.connect()
try:
return connection.info[key]
id(self),
)
if info:
- self.info = info # type: ignore
+ self.info = info # type: ignore[misc]
if (
attribute_options
# class, only on subclasses of it, which might be
# different. only return for the specific
# object's current value
- return inst._non_canonical_get_for_object(obj) # type: ignore
+ return inst._non_canonical_get_for_object(obj) # type: ignore[no-any-return] # noqa: E501
else:
- return inst # type: ignore # TODO
+ return inst # type: ignore[no-any-return] # TODO
def _calc_owner(self, target_cls: Any) -> Any:
# we might be getting invoked for a subclass
@property
def _comparator(self) -> PropComparator[Any]:
- return getattr( # type: ignore
+ return getattr( # type: ignore[no-any-return]
self.owning_class, self.target_collection
).comparator
@property
def _comparator(self) -> PropComparator[Any]:
- return getattr( # type: ignore
+ return getattr( # type: ignore[no-any-return]
self.aliased_insp.entity, self.target_collection
).comparator
# compatible with None.".
if key not in self.col:
self.col[key] = self._create(key, default)
- return default # type: ignore
+ return default # type: ignore[return-value]
else:
return self[key]
for member in removals:
remover(member)
- def __ior__( # type: ignore
+ def __ior__( # type: ignore[override]
self, other: AbstractSet[_S]
) -> MutableSet[Union[_T, _S]]:
if not collections._set_binops_check_strict(self, other):
for value in add:
self.add(value)
- def __ixor__(self, other: AbstractSet[_S]) -> MutableSet[Union[_T, _S]]: # type: ignore # noqa: E501
+ def __ixor__(self, other: AbstractSet[_S]) -> MutableSet[Union[_T, _S]]: # type: ignore[override] # noqa: E501
if not collections._set_binops_check_strict(self, other):
return NotImplemented
else:
proxy = proxy_ref()
if proxy is not None:
- return proxy # type: ignore
+ return proxy # type: ignore[return-value]
if regenerate:
return cls._regenerate_proxy_for_target(target, **additional_kw)
args: Tuple[Any, ...],
kwds: Dict[str, Any],
):
- self.gen = func(*args, **kwds) # type: ignore
+ self.gen = func(*args, **kwds) # type: ignore[assignment]
async def start(self, is_ctxmanager: bool = False) -> _T_co:
try:
# note that to send adapted arguments like
# prepared_statement_cache_size, user would use
# "creator" and emulate this form here
- return sync_engine.dialect.dbapi.connect( # type: ignore
+ return sync_engine.dialect.dbapi.connect( # type: ignore[union-attr] # noqa: E501
async_creator_fn=async_creator
)
workaround for SQLAlchemy 2.1.
"""
- return self # type: ignore
+ return self # type: ignore[return-value]
@deprecated(
"2.1.0",
"""
- return self # type: ignore
+ return self # type: ignore[return-value]
@_generative
def unique(self, strategy: Optional[_UniqueFilterType] = None) -> Self:
return result
if not is_cursor:
- cursor_result = getattr(result, "raw", None) # type: ignore
+ cursor_result = getattr(result, "raw", None) # type: ignore[assignment] # noqa: E501
else:
- cursor_result = result # type: ignore
+ cursor_result = result # type: ignore[assignment]
if cursor_result and cursor_result.context._is_server_side:
await greenlet_spawn(cursor_result.close)
raise async_exc.AsyncMethodRequired(
self,
bind: Optional[_AsyncSessionBind] = None,
*,
- class_: Type[_AS] = AsyncSession, # type: ignore
+ class_: Type[_AS] = AsyncSession, # type: ignore[assignment]
autoflush: bool = True,
expire_on_commit: bool = True,
info: Optional[_InfoType] = None,
await greenlet_spawn(_sync_close_all_sessions)
-_instance_state._async_provider = async_session # type: ignore
+_instance_state._async_provider = async_session # type: ignore[attr-defined]
if expr is not None:
self.expression(expr)
else:
- self.expression(func) # type: ignore
+ self.expression(func) # type: ignore[arg-type]
@property
def inplace(self) -> Self:
self, instance: Optional[object], owner: Type[object]
) -> Union[Callable[_P, _R], Callable[_P, SQLCoreOperations[_R]]]:
if instance is None:
- return self.expr.__get__(owner, owner) # type: ignore
+ return self.expr.__get__(owner, owner) # type: ignore[no-any-return] # noqa: E501
else:
- return self.func.__get__(instance, owner) # type: ignore
+ return self.func.__get__(instance, owner) # type: ignore[no-any-return] # noqa: E501
def expression(
self, expr: Callable[Concatenate[Any, _P], SQLCoreOperations[_R]]
def _unwrap_classmethod(meth: _T) -> _T:
if isinstance(meth, classmethod):
- return meth.__func__ # type: ignore
+ return meth.__func__ # type: ignore[return-value]
else:
return meth
# this accessor is not normally used, however is accessed by things
# like ORM synonyms if the hybrid is used in this context; the
# .property attribute is not necessarily accessible
- return self.expression.property # type: ignore
+ return self.expression.property # type: ignore[no-any-return, union-attr] # noqa: E501
def operate(
self, op: OperatorType, *other: Any, **kwargs: Any
def reverse_operate(
self, op: OperatorType, other: Any, **kwargs: Any
) -> ColumnElement[Any]:
- return op(other, self.expression, **kwargs) # type: ignore
+ return op(other, self.expression, **kwargs) # type: ignore[no-any-return] # noqa: E501
else:
name = _qual_logger_name_for_cls(instance.__class__)
- instance._echo = echoflag # type: ignore
+ instance._echo = echoflag # type: ignore[misc]
logger: Union[logging.Logger, InstanceLogger]
# levels by calling logger._log()
logger = InstanceLogger(echoflag, name)
- instance.logger = logger # type: ignore
+ instance.logger = logger # type: ignore[misc]
class echo_property:
def is_user_defined_option(
opt: ExecutableOption,
) -> TypeGuard[UserDefinedOption]:
- return not opt._is_core and opt._is_user_defined # type: ignore
+ return not opt._is_core and opt._is_user_defined # type: ignore[attr-defined] # noqa: E501
def is_composite_class(obj: Any) -> bool:
# interim class manager setup, there's a check for None to see if it
# needs to be populated, so we assign None here leaving the attribute
# in a temporarily not-type-correct state
- self.impl = impl # type: ignore
+ self.impl = impl # type: ignore[assignment]
assert comparator is not None
self.comparator = comparator
if key in base:
self.dispatch._update(base[key].dispatch)
if base[key].dispatch._active_history:
- self.dispatch._active_history = True # type: ignore
+ self.dispatch._active_history = True # type: ignore[attr-defined] # noqa: E501
_cache_key_traversal = [
("key", visitors.ExtendedInternalTraversal.dp_string),
def __doc__(self) -> Optional[str]:
return self._doc
- @__doc__.setter # type: ignore
+ @__doc__.setter # type: ignore[no-redef]
def __doc__(self, value: Optional[str]) -> None:
self._doc = value
- @__doc__.classlevel # type: ignore
+ @__doc__.classlevel # type: ignore[no-redef]
def __doc__(cls) -> Optional[str]:
return super().__doc__
class_, key, comparator=comparator, parententity=parententity
)
- descriptor.__doc__ = doc # type: ignore
+ descriptor.__doc__ = doc # type: ignore[method-assign]
manager.instrument_attribute(key, descriptor)
return descriptor
# can't get mypy to see an overload for this
insp = inspection.inspect(class_or_mapper, False)
if insp is not None:
- return insp.mapper # type: ignore
+ return insp.mapper # type: ignore[no-any-return]
else:
assert isinstance(class_or_mapper, type)
raise exc.UnmappedClassError(class_or_mapper)
# can't get mypy to see an overload for this
insp = inspection.inspect(entity, False)
if insp is not None:
- return insp.mapper # type: ignore
+ return insp.mapper # type: ignore[no-any-return]
else:
return None
self._raise_for_name(n.args[0], n)
-_fallback_dict: Mapping[str, Any] = None # type: ignore
+_fallback_dict: Mapping[str, Any] = None # type: ignore[assignment]
def _resolver(cls: Type[Any], prop: RelationshipProperty[Any]) -> Tuple[
# is realistically only during garbage collection of this object, so
# we type this as a callable that returns _AdaptedCollectionProtocol
# in all cases.
- self._data = weakref.ref(data) # type: ignore
+ self._data = weakref.ref(data) # type: ignore[assignment]
self.owner_state = owner_state
data._sa_adapter = self
self.owner_state = d["owner_state"]
# see note in constructor regarding this type: ignore
- self._data = weakref.ref(d["data"]) # type: ignore
+ self._data = weakref.ref(d["data"]) # type: ignore[assignment]
d["data"]._sa_adapter = self
self.invalidated = d["invalidated"]
@hybridproperty
def directive(cls) -> _declared_directive[Any]:
# see mapping_api.rst for docstring
- return _declared_directive # type: ignore
+ return _declared_directive # type: ignore[return-value]
@hybridproperty
def cascading(cls) -> _stateful_declared_attr[_T_co]:
_DeclarativeMapperConfig._assert_dc_arguments(current)
- cls_._sa_apply_dc_transforms = { # type: ignore # noqa: E501
+ cls_._sa_apply_dc_transforms = { # type: ignore[attr-defined] # noqa: E501
k: current.get(k, _NoArg.NO_ARG) if v is _NoArg.NO_ARG else v
for k, v in apply_dc_transforms.items()
}
# we search through full __mro__ for types. however...
sql_type = self.type_annotation_map.get(pt)
if sql_type is None:
- sql_type = sqltypes._type_map_get(pt) # type: ignore # noqa: E501
+ sql_type = sqltypes._type_map_get(pt) # type: ignore[arg-type] # noqa: E501
if sql_type is not None:
sql_type_inst = sqltypes.to_instance(sql_type)
def decorate(cls: Type[_T]) -> Type[_T]:
kw["cls"] = cls
kw["name"] = cls.__name__
- return self.generate_base(**kw) # type: ignore
+ return self.generate_base(**kw) # type: ignore[no-any-return]
return decorate
"""
_ORMClassConfigurator._as_declarative(self, cls, cls.__dict__)
- return cls.__mapper__ # type: ignore
+ return cls.__mapper__ # type: ignore[attr-defined, no-any-return]
def map_imperatively(
self,
continue
elif isinstance(value, Column):
_undefer_column_name(
- k, self.column_copies.get(value, value) # type: ignore
+ k, self.column_copies.get(value, value) # type: ignore[arg-type] # noqa: E501
)
else:
if isinstance(value, _IntrospectsAnnotations):
if hasattr(cls, "__table_cls__"):
table_cls = cast(
Type[Table],
- util.unbound_method_to_callable(cls.__table_cls__), # type: ignore # noqa: E501
+ util.unbound_method_to_callable(cls.__table_cls__), # type: ignore[no-untyped-call] # noqa: E501
)
else:
table_cls = Table
mapper_cls = cast(
"Type[Mapper[Any]]",
util.unbound_method_to_callable(
- self.cls.__mapper_cls__ # type: ignore
+ self.cls.__mapper_cls__ # type: ignore[no-untyped-call]
),
)
else:
@property
def cls(self) -> Type[Any]:
- return self._cls() # type: ignore
+ return self._cls() # type: ignore[return-value]
@cls.setter
def cls(self, class_: Type[Any]) -> None:
collection = False
@property
- def uses_objects(self) -> bool: # type: ignore
+ def uses_objects(self) -> bool: # type: ignore[override]
return prop.uses_objects
def __init__(self, key: str):
if isinstance(_class_or_attr, (Mapped, str, sql.ColumnElement)):
self.attrs = (_class_or_attr,) + attrs
# will initialize within declarative_scan
- self.composite_class = None # type: ignore
+ self.composite_class = None # type: ignore[assignment]
else:
- self.composite_class = _class_or_attr # type: ignore
+ self.composite_class = _class_or_attr # type: ignore[assignment]
self.attrs = attrs
self.return_none_on = return_none_on
" method; can't get state"
) from ae
else:
- return accessor() # type: ignore
+ return accessor() # type: ignore[no-any-return]
def do_init(self) -> None:
"""Initialization which occurs after the :class:`.Composite`
)
proxy_attr = self.parent.class_manager[self.key]
- proxy_attr.impl.dispatch = proxy_attr.dispatch # type: ignore
+ proxy_attr.impl.dispatch = proxy_attr.dispatch # type: ignore[assignment] # noqa: E501
proxy_attr.impl.dispatch._active_history = self.active_history
# TODO: need a deserialize hook here
else:
def get_values(val: Any) -> Tuple[Any]:
- return val.__composite_values__() # type: ignore
+ return val.__composite_values__() # type: ignore[no-any-return] # noqa: E501
attrs = [prop.key for prop in self.props]
"""
# https://github.com/python/mypy/issues/4266
- __hash__ = None # type: ignore
+ __hash__ = None # type: ignore[assignment]
prop: RODescriptorReference[Composite[_PT]]
comparator_callable = p.comparator_factory
break
assert comparator_callable is not None
- return comparator_callable(p, mapper) # type: ignore
+ return comparator_callable(p, mapper) # type: ignore[no-any-return]
def __init__(self) -> None:
super().__init__()
base = util.preloaded.orm_base
try:
- mappers = base.manager_of_class(cls).mappers # type: ignore
+ mappers = base.manager_of_class(cls).mappers # type: ignore[attr-defined] # noqa: E501
except (
UnmappedClassError,
TypeError,
self._wr = weakref.ref(self)
def _kill(self) -> None:
- self._add_unpresent = _killed # type: ignore
+ self._add_unpresent = _killed # type: ignore[method-assign]
def all_states(self) -> List[InstanceState[Any]]:
raise NotImplementedError()
self.uninstall_member(key)
self.mapper = None
- self.dispatch = None # type: ignore
+ self.dispatch = None # type: ignore[assignment]
self.new_init = None
self.info.clear()
"""
- return getattr(self.parent.class_, self.key) # type: ignore
+ return getattr(self.parent.class_, self.key) # type: ignore[no-any-return] # noqa: E501
def do_init(self) -> None:
"""Perform subclass-specific initialization post-mapper-creation
"""
- return self.operate(PropComparator.of_type_op, class_) # type: ignore
+ return self.operate(PropComparator.of_type_op, class_) # type: ignore[return-value] # noqa: E501
def and_(
self, *criteria: _ColumnExpressionArgument[bool]
:func:`.with_loader_criteria`
"""
- return self.operate(operators.and_, *criteria) # type: ignore
+ return self.operate(operators.and_, *criteria) # type: ignore[return-value] # noqa: E501
def any(
self,
if not isinstance(c.table, expression.TableClause):
return None
else:
- return c.table.key # type: ignore
+ return c.table.key # type: ignore[attr-defined, no-any-return]
colkeys = [(c.key, _table_key(c)) for c in cols]
return _SerializableColumnGetterV2, (colkeys,)
metadata = getattr(mapper.local_table, "metadata", None)
for ckey, tkey in self.colkeys:
if tkey is None or metadata is None or tkey not in metadata:
- cols.append(mapper.local_table.c[ckey]) # type: ignore
+ cols.append(mapper.local_table.c[ckey]) # type: ignore[index]
else:
cols.append(metadata.tables[tkey].c[ckey])
return cols
# interim - polymorphic_on is further refined in
# _configure_polymorphic_setter
self.polymorphic_on = (
- coercions.expect( # type: ignore
+ coercions.expect( # type: ignore[assignment]
roles.ColumnArgumentOrKeyRole,
polymorphic_on,
argname="polymorphic_on",
if fc.primary_key and pk_cols.issuperset(fc.primary_key):
# ordering is important since it determines the ordering of
# mapper.primary_key (and therefore query.get())
- self._pks_by_table[fc] = util.ordered_column_set( # type: ignore # noqa: E501
+ self._pks_by_table[fc] = util.ordered_column_set( # type: ignore[assignment] # noqa: E501
fc.primary_key
).intersection(
pk_cols
)
- self._cols_by_table[fc] = util.ordered_column_set(fc.c).intersection( # type: ignore # noqa: E501
+ self._cols_by_table[fc] = util.ordered_column_set(fc.c).intersection( # type: ignore[assignment] # noqa: E501
all_cols
)
"in properties.ColumnProperty %s",
key,
)
- return new_prop # type: ignore
+ return new_prop # type: ignore[no-any-return]
@util.preload_module("sqlalchemy.orm.descriptor_props")
def _property_from_column(
else:
return
- Mapper.dispatch._for_class(Mapper).before_configured() # type: ignore # noqa: E501
+ Mapper.dispatch._for_class(Mapper).before_configured() # type: ignore[arg-type, call-arg, misc] # noqa: E501
# initialize properties on all mappers
# note that _mapper_registry is unordered, which
_already_compiling = False
for reg in registries_configured:
reg.dispatch.after_configured(reg)
- Mapper.dispatch._for_class(Mapper).after_configured() # type: ignore
+ Mapper.dispatch._for_class(Mapper).after_configured() # type: ignore[arg-type, call-arg, misc] # noqa: E501
@util.preload_module("sqlalchemy.orm.decl_api")
"Original exception was: %s"
% (mapper, mapper._configure_failed)
)
- e._configure_failed = mapper._configure_failed # type: ignore
+ e._configure_failed = mapper._configure_failed # type: ignore[attr-defined] # noqa: E501
raise e
if not mapper.configured:
return self.path
def odd_element(self, index: int) -> _InternalEntityType[Any]:
- return self.path[index] # type: ignore
+ return self.path[index] # type: ignore[return-value]
def set(self, attributes: Dict[Any, Any], key: Any, value: Any) -> None:
log.debug("set '%s' on path '%s' to '%s'", key, self, value)
return prev[next_]
# can't quite get mypy to appreciate this one :)
- return reduce(_red, raw, cls.root) # type: ignore
+ return reduce(_red, raw, cls.root) # type: ignore[arg-type]
def __add__(self, other: PathRegistry) -> PathRegistry:
def _red(prev: PathRegistry, next_: _PathElementType) -> PathRegistry:
return _TokenRegistry(self, PathToken._intern[entity])
else:
try:
- return entity._path_registry # type: ignore
+ return entity._path_registry # type: ignore[no-any-return]
except AttributeError:
raise IndexError(
f"invalid argument for RootRegistry.__getitem__: {entity}"
parent.mapper.inherits
)
- if not insp.is_aliased_class or insp._use_mapper_path: # type: ignore
+ if not insp.is_aliased_class or insp._use_mapper_path: # type: ignore[union-attr] # noqa: E501
parent = natural_parent = parent.parent[prop.parent]
elif (
insp.is_aliased_class
and insp.with_polymorphic_mappers
and prop.parent in insp.with_polymorphic_mappers
):
- subclass_entity: _InternalEntityType[Any] = parent[-1]._entity_for_mapper(prop.parent) # type: ignore # noqa: E501
+ subclass_entity: _InternalEntityType[Any] = parent[-1]._entity_for_mapper(prop.parent) # type: ignore[union-attr] # noqa: E501
parent = parent.parent[subclass_entity]
# when building a path where with_polymorphic() is in use,
if earliest is None:
return self
else:
- return self.coerce(self.path[0 : -(earliest + 1)]) # type: ignore
+ return self.coerce(self.path[0 : -(earliest + 1)]) # type: ignore[return-value] # noqa: E501
@property
def entity_path(self) -> _AbstractEntityRegistry:
parent_natural_entity = parent.natural_path[-1]
if entity.mapper.isa(
- parent_natural_entity.mapper # type: ignore
- ) or parent_natural_entity.mapper.isa( # type: ignore
+ parent_natural_entity.mapper # type: ignore[union-attr]
+ ) or parent_natural_entity.mapper.isa( # type: ignore[union-attr]
entity.mapper
):
# when the entity mapper and parent mapper are in an
self.natural_path = parent.natural_path + (entity.mapper,)
else:
self.natural_path = parent.natural_path + (
- parent_natural_entity.entity, # type: ignore
+ parent_natural_entity.entity, # type: ignore[operator, union-attr] # noqa: E501
)
# it seems to make sense that since these paths get mixed up
# with statements that are cached or not, we should make
def columns_to_assign(self) -> List[Tuple[Column[Any], int]]:
# mypy doesn't care about the isinstance here
return [
- (c, 0) # type: ignore
+ (c, 0) # type: ignore[misc]
for c in self.columns
if isinstance(c, Column) and c.table is None
]
def _memoized_attr__renders_in_subqueries(self) -> bool:
if ("query_expression", True) in self.strategy_key:
- return self.strategy._have_default_expression # type: ignore
+ return self.strategy._have_default_expression # type: ignore[attr-defined, no-any-return] # noqa: E501
return ("deferred", True) not in self.strategy_key or (
self not in self.parent._readonly_props
ce = self.__clause_element__()
try:
- return ce.info # type: ignore
+ return ce.info # type: ignore[no-any-return]
except AttributeError:
return self.prop.info
# for the query(Entity).with_session(session) API which is likely in
# some old recipes, however these are legacy as select() can now be
# used.
- self.session = session # type: ignore
+ self.session = session # type: ignore[assignment]
self._set_entities(entities)
def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
:meth:`.Result.tuples` - v2 equivalent method.
"""
- return self.only_return_tuples(True) # type: ignore
+ return self.only_return_tuples(True) # type: ignore[return-value]
def _entity_from_pre_ent_zero(self) -> Optional[_InternalEntityType[Any]]:
if not self._raw_columns:
ent = self._raw_columns[0]
if "parententity" in ent._annotations:
- return ent._annotations["parententity"] # type: ignore
+ return ent._annotations["parententity"] # type: ignore[no-any-return] # noqa: E501
elif "bundle" in ent._annotations:
- return ent._annotations["bundle"] # type: ignore
+ return ent._annotations["bundle"] # type: ignore[no-any-return]
else:
# label, other SQL expression
for element in visitors.iterate(ent):
if "parententity" in element._annotations:
- return element._annotations["parententity"] # type: ignore # noqa: E501
+ return element._annotations["parententity"] # type: ignore[no-any-return] # noqa: E501
else:
return None
"a single mapped class." % methname
)
- return self._raw_columns[0]._annotations["parententity"] # type: ignore # noqa: E501
+ return self._raw_columns[0]._annotations["parententity"] # type: ignore[no-any-return] # noqa: E501
def _set_select_from(
self, obj: Iterable[_FromClauseArgument], set_base_alias: bool
return q._compile_state(
use_legacy_query_style=legacy_query_style
- ).statement # type: ignore
+ ).statement # type: ignore[return-value]
def _statement_20(
self, for_statement: bool = False, use_legacy_query_style: bool = True
new_query = fn(self)
if new_query is not None and new_query is not self:
self = new_query
- if not fn._bake_ok: # type: ignore
+ if not fn._bake_ok: # type: ignore[attr-defined]
self._compile_options += {"_bake_ok": False}
compile_options = self._compile_options
:attr:`.ORMExecuteState.lazy_loaded_from`
"""
- return self.load_options._lazy_loaded_from # type: ignore
+ return self.load_options._lazy_loaded_from # type: ignore[no-any-return] # noqa: E501
@property
def _current_path(self) -> PathRegistry:
- return self._compile_options._current_path # type: ignore
+ return self._compile_options._current_path # type: ignore[no-any-return] # noqa: E501
@_generative
def correlate(
for prop in mapper.iterate_properties:
if (
isinstance(prop, relationships.RelationshipProperty)
- and prop.mapper is entity_zero.mapper # type: ignore
+ and prop.mapper is entity_zero.mapper # type: ignore[union-attr] # noqa: E501
):
- property = prop # type: ignore # noqa: A001
+ property = prop # type: ignore[assignment] # noqa: A001
break
else:
raise sa_exc.InvalidRequestError(
"Could not locate a property which relates instances "
"of class '%s' to instances of class '%s'"
% (
- entity_zero.mapper.class_.__name__, # type: ignore
+ entity_zero.mapper.class_.__name__, # type: ignore[union-attr] # noqa: E501
instance.__class__.__name__,
)
)
return self.filter(
with_parent(
instance,
- property, # type: ignore
- entity_zero.entity, # type: ignore
+ property, # type: ignore[arg-type]
+ entity_zero.entity, # type: ignore[union-attr]
)
)
"""
try:
- return next(self._values_no_warn(column))[0] # type: ignore
+ return next(self._values_no_warn(column))[0] # type: ignore[arg-type, call-overload] # noqa: E501
except StopIteration:
return None
# Query has all the same fields as Select for this operation
# this could in theory be based on a protocol but not sure if it's
# worth it
- _MemoizedSelectEntities._generate_for_statement(self) # type: ignore
+ _MemoizedSelectEntities._generate_for_statement(self) # type: ignore[arg-type] # noqa: E501
self._set_entities(entities)
return self
if self._compile_options._current_path:
# opting for lower method overhead for the checks
for opt in opts:
- if not opt._is_core and opt._is_legacy_option: # type: ignore
- opt.process_query_conditionally(self) # type: ignore
+ if not opt._is_core and opt._is_legacy_option: # type: ignore[attr-defined] # noqa: E501
+ opt.process_query_conditionally(self) # type: ignore[attr-defined] # noqa: E501
else:
for opt in opts:
- if not opt._is_core and opt._is_legacy_option: # type: ignore
- opt.process_query(self) # type: ignore
+ if not opt._is_core and opt._is_legacy_option: # type: ignore[attr-defined] # noqa: E501
+ opt.process_query(self) # type: ignore[attr-defined]
self._with_options += opts
return self
:meth:`_engine.Result.scalars` - v2 comparable method.
"""
- return self._iter().all() # type: ignore
+ return self._iter().all() # type: ignore[return-value]
@_generative
@_assertions(_no_clauseelement_condition)
"""
# replicates limit(1) behavior
if self._statement is not None:
- return self._iter().first() # type: ignore
+ return self._iter().first() # type: ignore[return-value]
else:
- return self.limit(1)._iter().first() # type: ignore
+ return self.limit(1)._iter().first() # type: ignore[return-value]
def one_or_none(self) -> Optional[_T]:
"""Return at most one result or raise an exception.
:meth:`_engine.Result.scalar_one_or_none` - v2 comparable method.
"""
- return self._iter().one_or_none() # type: ignore
+ return self._iter().one_or_none() # type: ignore[return-value]
def one(self) -> _T:
"""Return exactly one result or raise an exception.
:meth:`_engine.Result.scalar_one` - v2 comparable method.
"""
- return self._iter().one() # type: ignore
+ return self._iter().one() # type: ignore[return-value]
def scalar(self) -> Any:
"""Return the first element of the first result or None
def __iter__(self) -> Iterator[_T]:
result = self._iter()
try:
- yield from result # type: ignore
+ yield from result # type: ignore[misc]
except GeneratorExit:
# issue #8710 - direct iteration is not reusable after
# an iterable block is broken, so close the result
# legacy: automatically set scalars, unique
if result._attributes.get("is_single_entity", False):
- result = result.scalars() # type: ignore
+ result = result.scalars() # type: ignore[assignment]
if result._attributes.get("filtered", False):
result = result.unique()
"""
col = sql.func.count(sql.literal_column("*"))
- return ( # type: ignore
+ return ( # type: ignore[no-any-return]
self._legacy_from_self(col).enable_eagerloads(False).scalar()
)
self = bulk_del.query
- delete_ = sql.delete(*self._raw_columns) # type: ignore
+ delete_ = sql.delete(*self._raw_columns) # type: ignore[arg-type]
if delete_args:
delete_ = delete_.with_dialect_options(**delete_args)
),
),
)
- bulk_del.result = result # type: ignore
+ bulk_del.result = result # type: ignore[attr-defined]
self.session.dispatch.after_bulk_delete(bulk_del)
result.close()
bulk_ud.query = new_query
self = bulk_ud.query
- upd = sql.update(*self._raw_columns) # type: ignore
+ upd = sql.update(*self._raw_columns) # type: ignore[arg-type]
ppo = update_args.pop("preserve_parameter_order", False)
if ppo:
- upd = upd.ordered_values(*values) # type: ignore
+ upd = upd.ordered_values(*values) # type: ignore[arg-type]
else:
upd = upd.values(values)
if update_args:
),
),
)
- bulk_ud.result = result # type: ignore
+ bulk_ud.result = result # type: ignore[attr-defined]
self.session.dispatch.after_bulk_update(bulk_ud)
result.close()
return result.rowcount
class RowReturningQuery(Query[Row[Unpack[_Ts]]]):
if TYPE_CHECKING:
- def tuples(self) -> Query[Tuple[Unpack[_Ts]]]: # type: ignore
+ def tuples(self) -> Query[Tuple[Unpack[_Ts]]]: # type: ignore[override] # noqa: E501
...
:func:`.foreign`
"""
- return _annotate_columns( # type: ignore
+ return _annotate_columns( # type: ignore[return-value]
coercions.expect(roles.ColumnArgumentRole, expr), {"remote": True}
)
"""
- return _annotate_columns( # type: ignore
+ return _annotate_columns( # type: ignore[return-value]
coercions.expect(roles.ColumnArgumentRole, expr), {"foreign": True}
)
attr_value = attr_value()
if isinstance(attr_value, attributes.QueryableAttribute):
- attr_value = attr_value.key # type: ignore
+ attr_value = attr_value.key # type: ignore[assignment]
self.resolved = attr_value
self._reverse_property: Set[RelationshipProperty[Any]] = set()
if overlaps:
- self._overlaps = set(re.split(r"\s*,\s*", overlaps)) # type: ignore # noqa: E501
+ self._overlaps = set(re.split(r"\s*,\s*", overlaps)) # type: ignore[assignment] # noqa: E501
else:
self._overlaps = ()
@property
def back_populates(self) -> str:
- return self._init_args.back_populates.effective_value() # type: ignore
+ return self._init_args.back_populates.effective_value() # type: ignore[no-any-return] # noqa: E501
@back_populates.setter
def back_populates(self, value: str) -> None:
def _memoized_attr_entity(self) -> _InternalEntityType[_PT]:
if self._of_type:
- return inspect(self._of_type) # type: ignore
+ return inspect(self._of_type) # type: ignore[return-value]
else:
return self.prop.entity
)
# https://github.com/python/mypy/issues/4266
- __hash__ = None # type: ignore
+ __hash__ = None # type: ignore[assignment]
def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
"""Implement the ``==`` operator.
elif not is_write_only and not is_dynamic:
self.uselist = False
- if argument.__args__: # type: ignore
+ if argument.__args__: # type: ignore[union-attr]
if isinstance(arg_origin, type) and issubclass(
arg_origin, typing.Mapping
):
- type_arg = argument.__args__[-1] # type: ignore
+ type_arg = argument.__args__[-1] # type: ignore[union-attr] # noqa: E501
else:
- type_arg = argument.__args__[0] # type: ignore
+ type_arg = argument.__args__[0] # type: ignore[union-attr]
if hasattr(type_arg, "__forward_arg__"):
str_argument = type_arg.__forward_arg__
try:
entity = inspect(resolved_argument)
except sa_exc.NoInspectionAvailable:
- entity = None # type: ignore
+ entity = None # type: ignore[assignment]
if not hasattr(entity, "mapper"):
raise sa_exc.ArgumentError(
if self.uselist is None:
self.uselist = self.direction is not MANYTOONE
if not self.viewonly:
- self._dependency_processor = ( # type: ignore
+ self._dependency_processor = ( # type: ignore[no-untyped-call]
dependency._DependencyProcessor.from_relationship
)(self)
def _annotate_columns(element: _CE, annotations: _AnnotationDict) -> _CE:
def clone(elem: _CE) -> _CE:
if isinstance(elem, expression.ColumnClause):
- elem = elem._annotate(annotations.copy()) # type: ignore
+ elem = elem._annotate(annotations.copy()) # type: ignore[attr-defined] # noqa: E501
elem._copy_internals(clone=clone)
return elem
if element is not None:
element = clone(element)
- clone = None # type: ignore # remove gc cycles
+ clone = None # type: ignore[assignment] # remove gc cycles
return element
def __get__(s, instance: Any, owner: Type[_O]) -> Query[_O]:
if query_cls:
# custom query class
- return query_cls(owner, session=self.registry()) # type: ignore # noqa: E501
+ return query_cls(owner, session=self.registry()) # type: ignore[return-value] # noqa: E501
else:
# session's configured query class
return self.registry().query(owner)
if opts is not None and opts.isinstance(
context._ORMCompileState.default_compile_options
):
- return opts # type: ignore
+ return opts # type: ignore[return-value]
else:
return None
if (
join_transaction_mode
and join_transaction_mode
- not in JoinTransactionMode.__args__ # type: ignore
+ not in JoinTransactionMode.__args__ # type: ignore[attr-defined]
):
raise sa_exc.ArgumentError(
f"invalid selection for join_transaction_mode: "
self,
bind: Optional[_SessionBind] = None,
*,
- class_: Type[_S] = Session, # type: ignore
+ class_: Type[_S] = Session, # type: ignore[assignment]
autoflush: bool = True,
expire_on_commit: bool = True,
info: Optional[_InfoType] = None,
del obj
self._cleanup(self.obj)
- self.obj = lambda: None # type: ignore
+ self.obj = lambda: None # type: ignore[assignment]
def _cleanup(self, ref: weakref.ref[_O]) -> None:
"""Weakref callback cleanup.
self.obj = weakref.ref(inst, self._cleanup)
self.class_ = inst.__class__
else:
- self.obj = lambda: None # type: ignore
+ self.obj = lambda: None # type: ignore[assignment]
self.class_ = state_dict["class_"]
self.committed_state = state_dict.get("committed_state", {})
return to_chop
elif (
c_token != f"{_RELATIONSHIP_TOKEN}:{_WILDCARD_TOKEN}"
- and c_token != p_token.key # type: ignore
+ and c_token != p_token.key # type: ignore[union-attr]
):
return None
)
if not path:
- return None # type: ignore
+ return None # type: ignore[return-value]
assert opt.is_token_strategy == path.is_token
# TODO: need to figure out this None thing being returned by
# inspect(), it should not have None as an option in most cases
# if at all
- insp: InspectionAttr = inspect(attr) # type: ignore
+ insp: InspectionAttr = inspect(attr) # type: ignore[assignment]
except sa_exc.NoInspectionAvailable as err:
raise sa_exc.ArgumentError(
"expected ORM mapped attribute for loader strategy argument"
cls, value_list: Optional[Union[Iterable[str], str]]
) -> CascadeOptions:
if isinstance(value_list, str) or value_list is None:
- return cls.from_string(value_list) # type: ignore
+ return cls.from_string(value_list) # type: ignore[no-any-return]
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
}
def __setstate__(self, state: Dict[str, Any]) -> None:
- self.__init__( # type: ignore
+ self.__init__( # type: ignore[misc]
state["entity"],
state["mapper"],
state["alias"],
self.where_criteria._resolve_with_args(ext_info.entity),
)
else:
- crit = self.where_criteria # type: ignore
+ crit = self.where_criteria # type: ignore[assignment]
assert isinstance(crit, ColumnElement)
return sql_util._deep_annotate(
crit,
if (
not prop
and getattr(right_info, "mapper", None)
- and right_info.mapper.single # type: ignore
+ and right_info.mapper.single # type: ignore[union-attr]
):
right_info = cast("_InternalEntityType[Any]", right_info)
# if single inheritance target and we are using a manual
"module level. See chained stack trace for more hints."
) from ce
except NameError as ne:
- if raiseerr and "Mapped[" in raw_annotation: # type: ignore
+ if raiseerr and "Mapped[" in raw_annotation: # type: ignore[operator]
raise orm_exc.MappedAnnotationError(
f"Could not interpret annotation {raw_annotation}. "
"Check that it uses names that are correctly imported at the "
"module level. See chained stack trace for more hints."
) from ne
- annotated = raw_annotation # type: ignore
+ annotated = raw_annotation # type: ignore[assignment]
if is_dataclass_field:
return annotated, None
# which actually started failing when pytest warnings plugin was
# turned on, due to util.warn() above
if fairy is not None:
- fairy.dbapi_connection = None # type: ignore
+ fairy.dbapi_connection = None # type: ignore[assignment]
fairy._connection_record = None
del dbapi_connection
del connection_record
if not soft:
# prevent any rollback / reset actions etc. on
# the connection
- self.dbapi_connection = None # type: ignore
+ self.dbapi_connection = None # type: ignore[assignment]
# finalize
self._checkin()
# can't get the descriptor assignment to work here
# in pylance. mypy is OK w/ it
- self.info = self.info.copy() # type: ignore
+ self.info = self.info.copy() # type: ignore[misc]
self._connection_record = None
"""
if isinstance(expression, operators.ColumnOperators):
- return expression.collate(collation) # type: ignore
+ return expression.collate(collation) # type: ignore[return-value]
else:
return CollationClause._create_collation_expression(
expression, collation
def is_insert_update(c: ClauseElement) -> TypeGuard[ValuesBase]:
- return c.is_dml and (c.is_insert or c.is_update) # type: ignore
+ return c.is_dml and (c.is_insert or c.is_update) # type: ignore[attr-defined] # noqa: E501
def _no_kw() -> exc.ArgumentError:
.. versionadded:: 2.0.20
"""
- return val # type: ignore
+ return val # type: ignore[return-value]
updated by the given dictionary.
"""
- return Annotated._as_annotated_instance(self, values) # type: ignore
+ return Annotated._as_annotated_instance(self, values) # type: ignore[return-value] # noqa: E501
def _with_annotations(self, values: _AnnotationDict) -> Self:
"""return a copy of this ClauseElement with annotations
replaced by the given dictionary.
"""
- return Annotated._as_annotated_instance(self, values) # type: ignore
+ return Annotated._as_annotated_instance(self, values) # type: ignore[return-value] # noqa: E501
@overload
def _deannotate(
if element is not None:
element = cast(_SA, clone(element))
- clone = None # type: ignore # remove gc cycles
+ clone = None # type: ignore[assignment] # remove gc cycles
return element
if element is not None:
element = cast(_SA, clone(element))
- clone = None # type: ignore # remove gc cycles
+ clone = None # type: ignore[assignment] # remove gc cycles
return element
# some classes include this even if they have traverse_internals
# e.g. BindParameter, add it if present.
if cls.__dict__.get("inherit_cache", False):
- anno_cls.inherit_cache = True # type: ignore
+ anno_cls.inherit_cache = True # type: ignore[attr-defined]
elif "inherit_cache" in cls.__dict__:
- anno_cls.inherit_cache = cls.__dict__["inherit_cache"] # type: ignore
+ anno_cls.inherit_cache = cls.__dict__["inherit_cache"] # type: ignore[attr-defined] # noqa: E501
anno_cls._is_column_operators = issubclass(cls, operators.ColumnOperators)
) -> _DefaultDescriptionTuple:
return (
_DefaultDescriptionTuple(
- default.arg, # type: ignore
+ default.arg, # type: ignore[attr-defined]
default.is_scalar,
default.is_callable,
default.is_sentinel,
@classmethod
def _create_singleton(cls) -> None:
obj = object.__new__(cls)
- obj.__init__() # type: ignore
+ obj.__init__() # type: ignore[misc]
# for a long time this was an empty frozenset, meaning
# a SingletonConstant would never be a "corresponding column" in
return self
decorated = _generative(fn)
- decorated.non_generative = fn # type: ignore
+ decorated.non_generative = fn # type: ignore[attr-defined]
return decorated
def _clone(self, **kw):
"""Create a shallow copy of this ExecutableOption."""
c = self.__class__.__new__(self.__class__)
- c.__dict__ = dict(self.__dict__) # type: ignore
+ c.__dict__ = dict(self.__dict__) # type: ignore[misc]
return c
)
# https://github.com/python/mypy/issues/4266
- __hash__: Optional[int] = None # type: ignore
+ __hash__: Optional[int] = None # type: ignore[assignment]
def contains_column(self, col: ColumnElement[Any]) -> bool:
"""Checks if a column object exists in this collection"""
colkey: _COLKEY
if key is None:
- colkey = column.key # type: ignore
+ colkey = column.key # type: ignore[assignment]
else:
colkey = key
def __setstate__(self, state: Dict[str, Any]) -> None:
parent = state["_parent"]
- self.__init__(parent) # type: ignore
+ self.__init__(parent) # type: ignore[misc]
def corresponding_column(
self, column: _COL, require_embedded: bool = False
if default is not NO_ARG:
return getattr(ns, key, default)
else:
- return getattr(ns, key) # type: ignore
+ return getattr(ns, key) # type: ignore[no-any-return]
except AttributeError as err:
raise exc.InvalidRequestError(
'Entity namespace for "%s" has no property "%s"' % (entity, key)
elif meth is ANON_NAME:
elements = util.preloaded.sql_elements
if isinstance(obj, elements._anonymous_label):
- obj = obj.apply_map(anon_map) # type: ignore
+ obj = obj.apply_map(anon_map) # type: ignore[arg-type]
result += (attrname, obj)
elif meth is CALL_GEN_CACHE_KEY:
result += (
# Join, Aliased, etc.
# see #8790
- if self._gen_static_annotations_cache_key: # type: ignore # noqa: E501
- result += self._annotations_cache_key # type: ignore # noqa: E501
+ if self._gen_static_annotations_cache_key: # type: ignore[attr-defined] # noqa: E501
+ result += self._annotations_cache_key # type: ignore[attr-defined] # noqa: E501
else:
- result += self._gen_annotations_cache_key(anon_map) # type: ignore # noqa: E501
+ result += self._gen_annotations_cache_key(anon_map) # type: ignore[attr-defined] # noqa: E501
elif (
meth is InternalTraversal.dp_clauseelement_list
),
)
else:
- result += meth( # type: ignore
+ result += meth( # type: ignore[misc, operator]
attrname, obj, self, anon_map, bindparams
)
return result
# with namedtuple
# can't use "if not TYPE_CHECKING" because mypy rejects it
# inside of a NamedTuple
- def __hash__(self) -> Optional[int]: # type: ignore
+ def __hash__(self) -> Optional[int]: # type: ignore[override]
"""CacheKey itself is not hashable - hash the .key portion"""
return None
):
# set to None to just mark the in positiontup, it will not
# be replaced below.
- param_pos[bind_name] = None # type: ignore
+ param_pos[bind_name] = None # type: ignore[assignment]
else:
ph = f"{self._numeric_binds_identifier_char}{num}"
num += 1
# mypy is not able to see the two value types as the above Union,
# it just sees "object". don't know how to resolve
return {
- key: value # type: ignore
+ key: value # type: ignore[misc]
for key, value in (
(
self.bind_names[bindparam],
) -> Union[str, Tuple[str, str]]:
str_key = c_key_role(key)
if hasattr(key, "table") and key.table in _et:
- return (key.table.name, str_key) # type: ignore
+ return (key.table.name, str_key) # type: ignore[union-attr]
else:
return str_key
col: ColumnClause[Any],
) -> Union[str, Tuple[str, str]]:
if col.table in _et:
- return (col.table.name, col.key) # type: ignore
+ return (col.table.name, col.key) # type: ignore[attr-defined]
else:
return col.key
_column_as_key = functools.partial(
coercions.expect_as_key, roles.DMLColumnRole
)
- _getattr_col_key = _col_bind_name = operator.attrgetter("key") # type: ignore # noqa: E501
+ _getattr_col_key = _col_bind_name = operator.attrgetter("key") # type: ignore[assignment] # noqa: E501
return _column_as_key, _getattr_col_key, _col_bind_name
param = _create_bind_param(
compiler, c, None, process=process, name=name, **kw
)
- compiler.insert_prefetch.append(c) # type: ignore
+ compiler.insert_prefetch.append(c) # type: ignore[attr-defined]
return param
param = _create_bind_param(
compiler, c, None, process=process, name=name, **kw
)
- compiler.update_prefetch.append(c) # type: ignore
+ compiler.update_prefetch.append(c) # type: ignore[attr-defined]
return param
self._gen_table()
@property
- def element(self): # type: ignore
+ def element(self): # type: ignore[override]
return self.table
def to_metadata(self, metadata: MetaData, table: Table) -> Self:
return None
else:
- _skip_fn = None # type: ignore
+ _skip_fn = None # type: ignore[assignment]
return [
t
def _default_compiler(self) -> SQLCompiler:
dialect = self._default_dialect()
- return dialect.statement_compiler(dialect, self) # type: ignore
+ return dialect.statement_compiler(dialect, self) # type: ignore[no-any-return] # noqa: E501
def _clone(self, **kw: Any) -> Self:
"""Create a shallow copy of this ClauseElement.
"""
if self.callable:
# TODO: set up protocol for bind parameter callable
- return self.callable() # type: ignore
+ return self.callable() # type: ignore[no-any-return]
else:
return self.value
@property
def comparator(self):
- return self.type.comparator_factory(self) # type: ignore
+ return self.type.comparator_factory(self) # type: ignore[arg-type]
def self_group(
self, against: Optional[OperatorType] = None
def comparator(self):
# TODO: this seems wrong, it seems like we might not
# be using this method.
- return self.type.comparator_factory(self) # type: ignore
+ return self.type.comparator_factory(self) # type: ignore[arg-type]
def self_group(
self, against: Optional[OperatorType] = None
for clause in clauses
)
self._is_implicitly_boolean = operators.is_boolean(self.operator)
- self.type = type_api.to_instance(type_) # type: ignore
+ self.type = type_api.to_instance(type_) # type: ignore[assignment]
@property
def _flattened_operator_clauses(
for to_flat in convert_clauses
)
- return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
+ return cls._construct_raw(operator, flattened_clauses) # type: ignore[arg-type] # noqa: E501
else:
assert lcc
# just one element. return it as a single boolean element,
# if type is None, we get NULLTYPE, which is our _T. But I don't
# know how to get the overloads to express that correctly
- self.type = type_api.to_instance(type_) # type: ignore
+ self.type = type_api.to_instance(type_) # type: ignore[assignment]
def _wraps_unnamed_column(self):
ungrouped = self.element._ungroup()
# if type is None, we get NULLTYPE, which is our _T. But I don't
# know how to get the overloads to express that correctly
- self.type = type_api.to_instance(type_) # type: ignore
+ self.type = type_api.to_instance(type_) # type: ignore[assignment]
self.negate = negate
self._is_implicitly_boolean = operators.is_boolean(operator)
# this is using the eq/ne operator given int hash values,
# rather than Operator, so that "bool" can be based on
# identity
- return self.operator(*self._orig) # type: ignore
+ return self.operator(*self._orig) # type: ignore[call-overload]
else:
raise TypeError("Boolean value of this clause is not defined")
self.element = element
# nulltype assignment issue
- self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
+ self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore[arg-type] # noqa: E501
self._propagate_attrs = element._propagate_attrs
def _with_binary_element_type(self, type_):
*criterion: _ColumnExpressionArgument[bool],
):
self.func = func
- self.filter.non_generative(self, *criterion) # type: ignore
+ self.filter.non_generative(self, *criterion) # type: ignore[attr-defined] # noqa: E501
@_generative
def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
# if type is None, we get NULLTYPE, which is our _T. But I don't
# know how to get the overloads to express that correctly
- self.type = type_api.to_instance(type_) # type: ignore
+ self.type = type_api.to_instance(type_) # type: ignore[assignment]
self.is_literal = is_literal
if not a.type._isnull:
return a.type
else:
- return type_api.NULLTYPE # type: ignore
+ return type_api.NULLTYPE # type: ignore[return-value]
def _corresponding_column_or_error(fromclause, column, require_embedded=False):
# if type is None, we get NULLTYPE, which is our _T. But I don't
# know how to get the overloads to express that correctly
- self.type = type_api.to_instance(type_) # type: ignore
+ self.type = type_api.to_instance(type_) # type: ignore[assignment]
class _FunctionGenerator:
# passthru __ attributes; fixes pydoc
if name.startswith("__"):
try:
- return self.__dict__[name] # type: ignore
+ return self.__dict__[name] # type: ignore[no-any-return]
except KeyError:
raise AttributeError(name)
# if type is None, we get NULLTYPE, which is our _T. But I don't
# know how to get the overloads to express that correctly
- self.type = type_api.to_instance(type_) # type: ignore
+ self.type = type_api.to_instance(type_) # type: ignore[assignment]
FunctionElement.__init__(self, *clauses)
)
)
- self.type = type_api.to_instance( # type: ignore
+ self.type = type_api.to_instance( # type: ignore[assignment]
kwargs.pop("type_", None) or getattr(self, "type", None)
)
-register_function("cast", Cast) # type: ignore
-register_function("extract", Extract) # type: ignore
+register_function("cast", Cast) # type: ignore[arg-type]
+register_function("extract", Extract) # type: ignore[arg-type]
class next_value(GenericFunction[int]):
seq, schema.Sequence
), "next_value() accepts a Sequence object as input."
self.sequence = seq
- self.type = sqltypes.to_instance( # type: ignore
+ self.type = sqltypes.to_instance( # type: ignore[assignment]
seq.data_type or getattr(self, "type", None)
)
*args: _ColumnExpressionOrLiteralArgument[Optional[_T]],
**kwargs: Any,
) -> None:
- super().__init__(*args, **kwargs) # type: ignore
+ super().__init__(*args, **kwargs) # type: ignore[arg-type]
class coalesce(ReturnTypeFromOptionalArgs[_T]):
for deferred_copy_internals in self._transforms:
expr = deferred_copy_internals(expr)
- return expr # type: ignore
+ return expr # type: ignore[no-any-return]
def _copy_internals(
self, clone=_clone, deferred_copy_internals=None, **kw
**kwargs: Any,
) -> Operators:
if hasattr(left, "__sa_operate__"):
- return left.operate(self, right, *other, **kwargs) # type: ignore
+ return left.operate(self, right, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
elif self.python_impl:
- return self.python_impl(left, right, *other, **kwargs) # type: ignore # noqa: E501
+ return self.python_impl(left, right, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
else:
raise exc.InvalidRequestError(
f"Custom operator {self.opstring!r} can't be used with "
PrimaryKeyConstraint(
_implicit_generated=True
)._set_parent_with_dispatch(self)
- self.foreign_keys = set() # type: ignore
+ self.foreign_keys = set() # type: ignore[misc]
self._extra_dependencies: Set[Table] = set()
if self.schema is not None:
self.fullname = "%s.%s" % (self.schema, self.name)
raise exc.ArgumentError(
"May not pass name positionally and as a keyword."
)
- name = l_args.pop(0) # type: ignore
+ name = l_args.pop(0) # type: ignore[assignment]
elif l_args[0] is None:
l_args.pop(0)
if l_args:
raise exc.ArgumentError(
"May not pass type_ positionally and as a keyword."
)
- type_ = l_args.pop(0) # type: ignore
+ type_ = l_args.pop(0) # type: ignore[assignment]
elif l_args[0] is None:
l_args.pop(0)
# name = None is expected to be an interim state
# note this use case is legacy now that ORM declarative has a
# dedicated "column" construct local to the ORM
- super().__init__(name, type_) # type: ignore
+ super().__init__(name, type_) # type: ignore[arg-type]
- self.key = key if key is not None else name # type: ignore
+ self.key = key if key is not None else name # type: ignore[assignment]
self.primary_key = primary_key
self._insert_sentinel = insert_sentinel
self._omit_from_statements = _omit_from_statements
primary_key.add(c)
if fk:
- foreign_keys.update(fk) # type: ignore
+ foreign_keys.update(fk) # type: ignore[arg-type]
return c.key, c
self.constraint = _constraint
# .parent is not Optional under normal use
- self.parent = None # type: ignore
+ self.parent = None # type: ignore[assignment]
self.use_alter = use_alter
self.name = name
# our column is a Column, and any subquery etc. proxying us
# would be doing so via another Column, so that's what would
# be returned here
- return table.columns.corresponding_column(self.column) # type: ignore
+ return table.columns.corresponding_column(self.column) # type: ignore[return-value] # noqa: E501
@util.memoized_property
def _column_tokens(self) -> Tuple[Optional[str], str, Optional[str]]:
self.parent._setup_on_memoized_fks(set_type)
- self.column = column # type: ignore
+ self.column = column # type: ignore[misc]
@util.ro_memoized_property
def column(self) -> Column[Any]:
try:
argspec = util.get_callable_argspec(fn, no_self=True)
except TypeError:
- return util.wrap_callable(lambda ctx: fn(), fn) # type: ignore
+ return util.wrap_callable(lambda ctx: fn(), fn) # type: ignore[call-arg, no-any-return, no-untyped-call] # noqa: E501
defaulted = argspec[3] is not None and len(argspec[3]) or 0
positionals = len(argspec[0]) - defaulted
if positionals == 0:
- return util.wrap_callable(lambda ctx: fn(), fn) # type: ignore
+ return util.wrap_callable(lambda ctx: fn(), fn) # type: ignore[call-arg, no-any-return, no-untyped-call] # noqa: E501
elif positionals == 1:
- return fn # type: ignore
+ return fn # type: ignore[return-value]
else:
raise exc.ArgumentError(
"ColumnDefault Python function takes zero or one "
if table.primary_key is not self:
table.constraints.discard(table.primary_key)
- table.primary_key = self # type: ignore
+ table.primary_key = self # type: ignore[misc]
table.constraints.add(self)
table_pks = [c for c in table.c if c.primary_key]
self._columns.extend(columns)
- PrimaryKeyConstraint._autoincrement_column._reset(self) # type: ignore
+ PrimaryKeyConstraint._autoincrement_column._reset(self) # type: ignore[attr-defined] # noqa: E501
self._set_parent_with_dispatch(self.table)
def _replace(self, col: Column[Any]) -> None:
- PrimaryKeyConstraint._autoincrement_column._reset(self) # type: ignore
+ PrimaryKeyConstraint._autoincrement_column._reset(self) # type: ignore[attr-defined] # noqa: E501
self._columns.replace(col)
self.dispatch._sa_event_column_added_to_pk_constraint(self, col)
"""
return ddl.sort_tables(
- sorted(self.tables.values(), key=lambda t: t.key) # type: ignore
+ sorted(self.tables.values(), key=lambda t: t.key) # type: ignore[attr-defined] # noqa: E501
)
# overload needed to work around mypy this mypy
At runtime returns self unchanged, without performing any validation.
"""
- return self # type: ignore
+ return self # type: ignore[return-value]
@overload
def select(
# assigning these three collections separately is not itself
# atomic, but greatly reduces the surface for problems
self._columns = _columns
- self.primary_key = primary_key # type: ignore
- self.foreign_keys = foreign_keys # type: ignore
+ self.primary_key = primary_key # type: ignore[misc]
+ self.foreign_keys = foreign_keys # type: ignore[assignment, misc]
@util.ro_non_memoized_property
def entity_namespace(self) -> _EntityNamespace:
)
)
columns._populate_separate_keys(
- (col._tq_key_label, col) for col in _columns # type: ignore
+ (col._tq_key_label, col) for col in _columns # type: ignore[misc]
)
foreign_keys.update(
- itertools.chain(*[col.foreign_keys for col in _columns]) # type: ignore # noqa: E501
+ itertools.chain(*[col.foreign_keys for col in _columns]) # type: ignore[arg-type] # noqa: E501
)
def _copy_internals(
@util.ro_non_memoized_property
def implicit_returning(self) -> bool:
- return self.element.implicit_returning # type: ignore
+ return self.element.implicit_returning # type: ignore[attr-defined, no-any-return] # noqa: E501
@property
def original(self) -> ReturnsRows:
super().__init__()
self.name = name
self._columns = DedupeColumnCollection() # type: ignore[unused-ignore]
- self.primary_key = ColumnSet() # type: ignore
- self.foreign_keys = set() # type: ignore
+ self.primary_key = ColumnSet() # type: ignore[misc]
+ self.foreign_keys = set() # type: ignore[misc]
for c in columns:
self.append_column(c)
if c._allow_label_resolve
}
only_froms: Dict[str, ColumnElement[Any]] = {
- c.key: c # type: ignore
+ c.key: c # type: ignore[misc]
for c in _select_iterables(self.froms)
if c._allow_label_resolve
}
self._assert_no_memoizations()
if maintain_column_froms:
- self.select_from.non_generative( # type: ignore
+ self.select_from.non_generative( # type: ignore[attr-defined]
self, *self.columns_clause_froms
)
# cases. in other cases such as plain visitors.cloned_traverse(), we
# expect this to happen. see issue #12915
if not _annotations_traversal:
- ee = self._Annotated__element # type: ignore
+ ee = self._Annotated__element # type: ignore[attr-defined]
ee._copy_internals(**kw)
if ind_cols_on_fromclause:
# passed from annotations._deep_annotate(). See that function
# for notes
- ee = self._Annotated__element # type: ignore
- self.c = ee.__class__.c.fget(self) # type: ignore
+ ee = self._Annotated__element # type: ignore[attr-defined]
+ self.c = ee.__class__.c.fget(self) # type: ignore[misc]
@util.ro_memoized_property
def c(self) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]:
See test_selectable->test_annotated_corresponding_column
"""
- ee = self._Annotated__element # type: ignore
- return ee.c # type: ignore
+ ee = self._Annotated__element # type: ignore[attr-defined]
+ return ee.c # type: ignore[no-any-return]
kw["length"] = NO_ARG if self.length == 0 else self.length
return cast(
Enum,
- self._generic_type_affinity(_enums=enum_args, **kw), # type: ignore # noqa: E501
+ self._generic_type_affinity(_enums=enum_args, **kw), # type: ignore[call-arg] # noqa: E501
)
def _setup_for_values(
if impl:
# custom impl is not necessarily a LargeBinary subclass.
# make an exception to typing for this
- self.impl = to_instance(impl) # type: ignore
+ self.impl = to_instance(impl) # type: ignore[assignment]
def __reduce__(self):
return PickleType, (self.protocol, None, self.comparator)
def process(value: Any) -> Optional[dt.timedelta]:
if value is None:
return None
- return value - epoch # type: ignore
+ return value - epoch # type: ignore[no-any-return]
return process
"_generated_shallow_from_dict_traversal",
)
- cls._generated_shallow_from_dict_traversal = shallow_from_dict # type: ignore # noqa: E501
+ cls._generated_shallow_from_dict_traversal = shallow_from_dict # type: ignore[assignment, method-assign] # noqa: E501
shallow_from_dict(self, d)
cls._traverse_internals, "_generated_shallow_to_dict_traversal"
)
- cls._generated_shallow_to_dict_traversal = shallow_to_dict # type: ignore # noqa: E501
+ cls._generated_shallow_to_dict_traversal = shallow_to_dict # type: ignore[assignment, method-assign] # noqa: E501
return shallow_to_dict(self)
def _shallow_copy_to(self, other: Self) -> None:
cls._traverse_internals, "_generated_shallow_copy_traversal"
)
- cls._generated_shallow_copy_traversal = shallow_copy # type: ignore # noqa: E501
+ cls._generated_shallow_copy_traversal = shallow_copy # type: ignore[assignment, method-assign] # noqa: E501
shallow_copy(self, other)
def _clone(self, **kw: Any) -> Self:
# dmypy / mypy seems to sporadically keep thinking this line is
# returning Any, which seems to be caused by the @deprecated_params
# decorator on the DefaultDialect constructor
- return default.StrCompileDialect() # type: ignore
+ return default.StrCompileDialect() # type: ignore[no-any-return]
def __str__(self) -> str:
return str(self.compile())
"""
# dmypy seems to crash on this
- return cls(**kw) # type: ignore
+ return cls(**kw) # type: ignore[return-value]
# dmypy seems to crash with this, on repeated runs with changes
# if TYPE_CHECKING:
# impl_instance.
@util.memoized_property
def impl_instance(self) -> TypeEngine[Any]:
- return self.impl # type: ignore
+ return self.impl # type: ignore[return-value]
def __init__(self, *args: Any, **kwargs: Any):
"""Construct a :class:`.TypeDecorator`.
return type(
"TDComparator",
- (TypeDecorator.Comparator, impl.comparator_factory), # type: ignore # noqa: E501
+ (TypeDecorator.Comparator, impl.comparator_factory), # type: ignore[arg-type, return-value] # noqa: E501
{"__reduce__": __reduce__},
)
@property
- def comparator_factory( # type: ignore # mypy properties bug
+ def comparator_factory( # type: ignore[override] # mypy properties bug
self,
) -> _ComparatorFactory[Any]:
- if TypeDecorator.Comparator in self.impl.comparator_factory.__mro__: # type: ignore # noqa: E501
+ if TypeDecorator.Comparator in self.impl.comparator_factory.__mro__: # type: ignore[attr-defined] # noqa: E501
return self.impl_instance.comparator_factory
else:
# reconcile the Comparator class on the impl with that
# mypy property bug
@property
- def sort_key_function(self) -> Optional[Callable[[Any], Any]]: # type: ignore # noqa: E501
+ def sort_key_function(self) -> Optional[Callable[[Any], Any]]: # type: ignore[override] # noqa: E501
return self.impl_instance.sort_key_function
def repr_struct(self) -> util.GenericRepr:
# those are just column elements by themselves
yield element
elif element.__visit_name__ == "binary" and operators.is_comparison(
- element.operator # type: ignore
+ element.operator # type: ignore[attr-defined]
):
- stack.insert(0, element) # type: ignore
- for l in visit(element.left): # type: ignore
- for r in visit(element.right): # type: ignore
+ stack.insert(0, element) # type: ignore[arg-type]
+ for l in visit(element.left): # type: ignore[attr-defined]
+ for r in visit(element.right): # type: ignore[attr-defined]
fn(stack[0], l, r)
stack.pop(0)
for elem in element.get_children():
yield from visit(elem)
list(visit(expr))
- visit = None # type: ignore # remove gc cycles
+ visit = None # type: ignore[assignment] # remove gc cycles
def find_tables(
t = stack.popleft()
if isinstance(t, ColumnElement) and (
not isinstance(t, UnaryExpression)
- or not operators.is_ordering_modifier(t.modifier) # type: ignore
+ or not operators.is_ordering_modifier(t.modifier) # type: ignore[arg-type] # noqa: E501
):
if isinstance(t, Label) and not isinstance(
t.element, ScalarSelect
column_set = util.OrderedSet(columns)
cset_no_text: util.OrderedSet[ColumnElement[Any]] = column_set.difference(
- c for c in column_set if is_text_clause(c) # type: ignore
+ c for c in column_set if is_text_clause(c) # type: ignore[assignment]
)
omit = util.column_set()
# TODO: cython candidate
- if self.include_fn and not self.include_fn(col): # type: ignore
+ if self.include_fn and not self.include_fn(col): # type: ignore[arg-type] # noqa: E501
return None
- elif self.exclude_fn and self.exclude_fn(col): # type: ignore
+ elif self.exclude_fn and self.exclude_fn(col): # type: ignore[arg-type] # noqa: E501
return None
if isinstance(col, FromClause) and not isinstance(
break
else:
return None
- return self.selectable # type: ignore
+ return self.selectable # type: ignore[return-value]
elif isinstance(col, Alias) and isinstance(
col.element, TableClause
):
if TYPE_CHECKING:
assert isinstance(col, KeyedColumnElement)
- return self._corresponding_column( # type: ignore
+ return self._corresponding_column( # type: ignore[no-any-return]
col, require_embedded=True
)
adapt_from_selectables=adapt_from_selectables,
)
- self.columns = util.WeakPopulateDict(self._locate_col) # type: ignore
+ self.columns = util.WeakPopulateDict(self._locate_col) # type: ignore[arg-type, assignment] # noqa: E501
if self.include_fn or self.exclude_fn:
self.columns = self._IncludeExcludeMapping(self, self.columns)
self.adapt_required = adapt_required
def wrap(self, adapter):
ac = copy.copy(self)
ac._wrap = adapter
- ac.columns = util.WeakPopulateDict(ac._locate_col) # type: ignore
+ ac.columns = util.WeakPopulateDict(ac._locate_col) # type: ignore[arg-type, assignment] # noqa: E501
if ac.include_fn or ac.exclude_fn:
ac.columns = self._IncludeExcludeMapping(ac, ac.columns)
offset_clause = 0
if start != 0:
- offset_clause = offset_clause + start # type: ignore
+ offset_clause = offset_clause + start # type: ignore[operator]
if offset_clause == 0:
offset_clause = None
try:
meth = getter(visitor)
except AttributeError as err:
- return visitor.visit_unsupported_compilation(self, err, **kw) # type: ignore # noqa: E501
+ return visitor.visit_unsupported_compilation(self, err, **kw) # type: ignore[no-any-return] # noqa: E501
else:
- return meth(self, **kw) # type: ignore # noqa: E501
+ return meth(self, **kw) # type: ignore[no-any-return] # noqa: E501
- cls._compiler_dispatch = ( # type: ignore
+ cls._compiler_dispatch = ( # type: ignore[method-assign]
cls._original_compiler_dispatch
) = _compiler_dispatch
"""
name = _dispatch_lookup[visit_symbol]
- return getattr(self, name, None) # type: ignore
+ return getattr(self, name, None) # type: ignore[return-value]
def run_generated_dispatch(
self,
newelem = replace(elem)
if newelem is not None:
stop_on.add(id(newelem))
- return newelem # type: ignore
+ return newelem # type: ignore[no-any-return]
else:
# base "already seen" on id(), not hash, so that we don't
# replace an Annotated element with its non-annotated one, and
newelem = kw["replace"](elem)
if newelem is not None:
cloned[id_elem] = newelem
- return newelem # type: ignore
+ return newelem # type: ignore[no-any-return]
cloned[id_elem] = newelem = elem._clone(**kw)
newelem._copy_internals(clone=clone, **kw)
- return cloned[id_elem] # type: ignore
+ return cloned[id_elem] # type: ignore[no-any-return]
if obj is not None:
obj = clone(
def to_list(x: Any, default: Optional[List[Any]] = None) -> List[Any]:
if x is None:
- return default # type: ignore
+ return default # type: ignore[return-value]
if not is_non_string_iterable(x):
return [x]
elif isinstance(x, list):
while len(self) > self.capacity + self.capacity * self.threshold:
if size_alert:
size_alert = False
- self.size_alert(self) # type: ignore
+ self.size_alert(self) # type: ignore[misc]
by_counter = sorted(
self._data.values(),
key=operator.itemgetter(2),
decorated = warned(func)
decorated.__doc__ = doc
- decorated._sa_warn = lambda: _warn_with_version( # type: ignore
+ decorated._sa_warn = lambda: _warn_with_version( # type: ignore[attr-defined] # noqa: E501
message, version, wtype, stacklevel=3
)
return decorated
_exec_code_in_env(code, env, fn.__name__),
)
decorated.__defaults__ = fn.__defaults__
- decorated.__kwdefaults__ = fn.__kwdefaults__ # type: ignore
+ decorated.__kwdefaults__ = fn.__kwdefaults__ # type: ignore[union-attr] # noqa: E501
return update_wrapper(decorated, fn) # type: ignore[return-value]
return update_wrapper(decorate, target) # type: ignore[return-value]
self.__dict__[fn.__name__] = memo
return result
- return update_wrapper(oneshot, fn) # type: ignore
+ return update_wrapper(oneshot, fn) # type: ignore[return-value]
class HasMemoized:
self._memoized_keys |= {fn.__name__}
return result
- return update_wrapper(oneshot, fn) # type: ignore
+ return update_wrapper(oneshot, fn) # type: ignore[return-value]
if TYPE_CHECKING:
):
return set
else:
- return specimen.__emulates__ # type: ignore
+ return specimen.__emulates__ # type: ignore[no-any-return]
isa = issubclass if isinstance(specimen, type) else isinstance
if isa(specimen, list):
# will always be Type.
# the element here will be either ForwardRef or
# Optional[ForwardRef]
- return original_annotation # type: ignore
+ return original_annotation # type: ignore[return-value]
else:
_already_seen.add(annotation)
return _copy_generic_annotation_with(annotation, elements)
- return annotation # type: ignore
+ return annotation # type: ignore[return-value]
def fixup_container_fwd_refs(
)
):
# compat with py3.10 and earlier
- return get_origin(type_).__class_getitem__( # type: ignore
+ return get_origin(type_).__class_getitem__( # type: ignore[no-any-return, union-attr] # noqa: E501
tuple(
[
ForwardRef(elem) if isinstance(elem, str) else elem
) -> Type[_T]:
if hasattr(annotation, "copy_with"):
# List, Dict, etc. real generics
- return annotation.copy_with(elements) # type: ignore
+ return annotation.copy_with(elements) # type: ignore[no-any-return]
else:
# Python builtins list, dict, etc.
- return annotation.__origin__[elements] # type: ignore
+ return annotation.__origin__[elements] # type: ignore[no-any-return]
def eval_expression(
def make_union_type(*types: _AnnotationScanType) -> Type[Any]:
"""Make a Union type."""
- return Union[types] # type: ignore
+ return Union[types] # type: ignore[return-value]
def includes_none(type_: Any) -> bool:
[project.optional-dependencies]
asyncio = ["greenlet>=1"]
mypy = [
- "mypy >= 2",
+ "mypy >= 2.3",
"types-greenlet >= 2",
]
mssql = ["pyodbc"]
mypy_path = "./lib/"
show_error_codes = true
incremental = true
-# would be nice to enable this but too many error are surfaceds
-# enable_error_code = "ignore-without-code"
+enable_error_code = "ignore-without-code"
[[tool.mypy.overrides]]
"""
underscore_args = [arg.replace("-", "_") for arg in args]
- return_tuple = collections.namedtuple("options", underscore_args) # type: ignore # noqa: E501
+ return_tuple = collections.namedtuple("options", underscore_args) # type: ignore[misc] # noqa: E501
look_for_args = {f"--{arg}": idx for idx, arg in enumerate(args)}
return_args = [False for arg in args]
[testenv:pep484]
deps=
greenlet >= 1
- mypy >= 2
+ mypy >= 2.3
types-greenlet
commands =
mypy {env:MYPY_COLOR} ./lib/sqlalchemy
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
