driver = "aiomysql"
supports_statement_cache = True
- supports_server_side_cursors = True # type: ignore[assignment]
+ supports_server_side_cursors = True
_sscursor = AsyncAdapt_aiomysql_ss_cursor
is_async = True
driver = "asyncmy"
supports_statement_cache = True
- supports_server_side_cursors = True # type: ignore[assignment]
+ supports_server_side_cursors = True
_sscursor = AsyncAdapt_asyncmy_ss_cursor
is_async = True
return value
-class MySQLIdentifierPreparer_mysqlconnector( # type:ignore[misc]
+class MySQLIdentifierPreparer_mysqlconnector(
IdentifierPreparerCommon_mysqlconnector, MySQLIdentifierPreparer
):
pass
-class MariaDBIdentifierPreparer_mysqlconnector( # type:ignore[misc]
+class MariaDBIdentifierPreparer_mysqlconnector(
IdentifierPreparerCommon_mysqlconnector, MariaDBIdentifierPreparer
):
pass
return (0, 0, 0)
@util.langhelpers.memoized_property
- def supports_server_side_cursors(self) -> bool: # type: ignore[override]
+ def supports_server_side_cursors(self) -> bool:
try:
cursors = __import__("MySQLdb.cursors").cursors
self._sscursor = cursors.SSCursor
description_encoding = None
@langhelpers.memoized_property
- def supports_server_side_cursors(self) -> bool: # type: ignore[override]
+ def supports_server_side_cursors(self) -> bool:
try:
cursors = __import__("pymysql.cursors").cursors
self._sscursor = cursors.SSCursor
not insp.defaults or insp.defaults[0] is not False
)
- def do_ping(self, dbapi_connection: DBAPIConnection) -> Literal[True]: # type: ignore # noqa: E501
+ def do_ping(self, dbapi_connection: DBAPIConnection) -> Literal[True]:
if self._send_false_to_ping:
dbapi_connection.ping(False)
else:
def _against_native_enum(self) -> bool:
return (
isinstance(self.item_type, sqltypes.Enum)
- and self.item_type.native_enum # type: ignore[attr-defined]
+ and self.item_type.native_enum
)
def literal_processor(
value2 += step
value2_inc = False
- if value1 < value2: # type: ignore
+ if value1 < value2:
return -1
- elif value1 > value2: # type: ignore
+ elif value1 > value2:
return 1
elif only_values:
return 0
def _is_compiled() -> bool:
"""Utility function to indicate if this module is compiled or not."""
- return cython.compiled # type: ignore[no-any-return]
+ return cython.compiled # type: ignore[no-any-return,unused-ignore]
# END GENERATED CYTHON IMPORT
def _is_compiled() -> bool:
"""Utility function to indicate if this module is compiled or not."""
- return cython.compiled # type: ignore[no-any-return]
+ return cython.compiled # type: ignore[no-any-return,unused-ignore]
# END GENERATED CYTHON IMPORT
def _is_compiled() -> bool:
"""Utility function to indicate if this module is compiled or not."""
- return cython.compiled # type: ignore[no-any-return]
+ return cython.compiled # type: ignore[no-any-return,unused-ignore]
# END GENERATED CYTHON IMPORT
self._we_dont_return_rows()
@property
- def _keymap(self):
+ def _keymap(self): # type: ignore[override]
self._we_dont_return_rows()
@property
- def _key_to_index(self):
+ def _key_to_index(self): # type: ignore[override]
self._we_dont_return_rows()
@property
- def _processors(self):
+ def _processors(self): # type: ignore[override]
self._we_dont_return_rows()
@property
return weakref.WeakKeyDictionary()
@property
- def dialect_description(self):
+ def dialect_description(self): # type: ignore[override]
return self.name + "+" + self.driver
@property
return "unknown"
@property
- def executemany(self):
+ def executemany(self): # type: ignore[override]
return self.execute_style in (
ExecuteStyle.EXECUTEMANY,
ExecuteStyle.INSERTMANYVALUES,
if self._rowcount is None and exec_opt.get("preserve_rowcount", False):
self._rowcount = self.cursor.rowcount
+ yp: Optional[Union[int, bool]]
if self.is_crud or self.is_text:
result = self._setup_dml_or_text_result()
yp = False
if val is not None:
if coerce:
val = cls.coerce(key, val)
+ assert val is not None
state.dict[key] = val
val._parents[state] = key
self._doc = self.__doc__ = doc
@property
- def _parententity(self):
+ def _parententity(self): # type: ignore[override]
return inspection.inspect(self.class_, raiseerr=False)
@property
- def parent(self):
+ def parent(self): # type: ignore[override]
return inspection.inspect(self.class_, raiseerr=False)
_is_internal_proxy = True
def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
pass
- # mypy disallows plain property override of variable
- @property # type: ignore
+ @property
def cls(self) -> Type[Any]:
return self._cls() # type: ignore
"""
- primary_key: Tuple[Column[Any], ...]
+ primary_key: Tuple[ColumnElement[Any], ...]
"""An iterable containing the collection of :class:`_schema.Column`
objects
which comprise the 'primary key' of the mapped table, from the
if spec == "*":
mappers = list(self.self_and_descendants)
elif spec:
- mapper_set = set()
+ mapper_set: Set[Mapper[Any]] = set()
for m in util.to_list(spec):
m = _class_to_mapper(m)
if not m.isa(self):
return self.class_manager.mapper.base_mapper
def _result_has_identity_key(self, result, adapter=None):
- pk_cols: Sequence[ColumnClause[Any]] = self.primary_key
- if adapter:
- pk_cols = [adapter.columns[c] for c in pk_cols]
+ pk_cols: Sequence[ColumnElement[Any]]
+ if adapter is not None:
+ pk_cols = [adapter.columns[c] for c in self.primary_key]
+ else:
+ pk_cols = self.primary_key
rk = result.keys()
for col in pk_cols:
if col not in rk:
for the "row" argument
"""
- pk_cols: Sequence[ColumnClause[Any]] = self.primary_key
- if adapter:
- pk_cols = [adapter.columns[c] for c in pk_cols]
+ pk_cols: Sequence[ColumnElement[Any]]
+ if adapter is not None:
+ pk_cols = [adapter.columns[c] for c in self.primary_key]
+ else:
+ pk_cols = self.primary_key
mapping: RowMapping
if hasattr(row, "_mapping"):
return self.strategy._have_default_expression # type: ignore
return ("deferred", True) not in self.strategy_key or (
- self not in self.parent._readonly_props # type: ignore
+ self not in self.parent._readonly_props
)
@util.preload_module("sqlalchemy.orm.state", "sqlalchemy.orm.strategies")
else:
self._overlaps = ()
- # mypy ignoring the @property setter
- self.cascade = cascade # type: ignore
+ self.cascade = cascade
if back_populates:
if backref:
_propagate_attrs: _PropagateAttrsType = util.immutabledict()
- proxy_set = util.EMPTY_SET # type: ignore
+ proxy_set = util.EMPTY_SET
exprs: List[_ColumnsClauseElement]
return _DynamicCollectionAdapter(data) # type: ignore[return-value]
@util.memoized_property
- def _append_token( # type:ignore[override]
- self,
- ) -> attributes.AttributeEventToken:
+ def _append_token(self) -> attributes.AttributeEventToken:
return attributes.AttributeEventToken(self, attributes.OP_APPEND)
@util.memoized_property
- def _remove_token( # type:ignore[override]
- self,
- ) -> attributes.AttributeEventToken:
+ def _remove_token(self) -> attributes.AttributeEventToken:
return attributes.AttributeEventToken(self, attributes.OP_REMOVE)
def fire_append_event(
"""
- _is_asyncio = False # type: ignore[assignment]
+ _is_asyncio = False
_queue_class: Type[sqla_queue.QueueCommon[ConnectionPoolEntry]] = (
sqla_queue.Queue
"""
- _is_asyncio = True # type: ignore[assignment]
+ _is_asyncio = True
_queue_class: Type[sqla_queue.QueueCommon[ConnectionPoolEntry]] = (
sqla_queue.AsyncAdaptedQueue
)
"""
- _is_asyncio = False # type: ignore[assignment]
+ _is_asyncio = False
def __init__(
self,
def _is_compiled() -> bool:
"""Utility function to indicate if this module is compiled or not."""
- return cython.compiled # type: ignore[no-any-return]
+ return cython.compiled # type: ignore[no-any-return,unused-ignore]
# END GENERATED CYTHON IMPORT
)
@util.preload_module("sqlalchemy.sql.elements")
- def _literal_coercion(self, element, *, expr, operator, **kw):
+ def _literal_coercion(self, element, *, expr, operator, **kw): # type: ignore[override] # noqa: E501
if util.is_non_string_iterable(element):
non_literal_expressions: Dict[
Optional[_ColumnExpressionArgument[Any]],
if self.preparer._requires_quotes(cte_name):
cte_name = self.preparer.quote(cte_name)
text += self.get_render_as_alias_suffix(cte_name)
- return text
+ return text # type: ignore[no-any-return]
else:
return self.preparer.format_alias(cte, cte_name)
self.stack.pop(-1)
- return text
+ return text # type: ignore[no-any-return]
def delete_extra_from_clause(
self, delete_stmt, from_table, extra_froms, from_hints, **kw
self._ddl_if = getattr(element, "_ddl_if", None)
@property
- def stringify_dialect(self):
+ def stringify_dialect(self): # type: ignore[override]
assert not isinstance(self.element, str)
return self.element.create_drop_stringify_dialect
return self
@property
- def _constructor(self):
+ def _constructor(self): # type: ignore[override]
"""return the 'constructor' for this ClauseElement.
This is for the purposes for creating a new object of
else:
elem_cache_key = None
+ extracted_params: Optional[Sequence[BindParameter[Any]]]
if elem_cache_key is not None:
if TYPE_CHECKING:
assert compiled_cache is not None
_allow_label_resolve = False
@property
- def _is_star(self):
+ def _is_star(self): # type: ignore[override]
return self.text == "*"
def __init__(self, text: str):
return self
@property
- def primary_key(self):
+ def primary_key(self): # type: ignore[override]
return self.element.primary_key
@property
- def foreign_keys(self):
+ def foreign_keys(self): # type: ignore[override]
return self.element.foreign_keys
def _copy_internals(
_is_multiparam_column = False
@property
- def _is_star(self):
+ def _is_star(self): # type: ignore[override]
return self.is_literal and self.name == "*"
def __init__(
while lambda_element is not None:
rec = lambda_element._rec
if rec.bindparam_trackers:
- tracker_instrumented_fn = rec.tracker_instrumented_fn
+ tracker_instrumented_fn = (
+ rec.tracker_instrumented_fn # type:ignore [union-attr] # noqa: E501
+ )
for tracker in rec.bindparam_trackers:
tracker(
lambda_element.fn,
return self._rec_expected_expr
@property
- def _with_options(self):
+ def _with_options(self): # type: ignore[override]
return self._proxied._with_options
@property
return self._proxied._effective_plugin_target
@property
- def _execution_options(self):
+ def _execution_options(self): # type: ignore[override]
return self._proxied._execution_options
@property
return self._proxied._all_selected_columns
@property
- def is_select(self):
+ def is_select(self): # type: ignore[override]
return self._proxied.is_select
@property
- def is_update(self):
+ def is_update(self): # type: ignore[override]
return self._proxied.is_update
@property
- def is_insert(self):
+ def is_insert(self): # type: ignore[override]
return self._proxied.is_insert
@property
- def is_text(self):
+ def is_text(self): # type: ignore[override]
return self._proxied.is_text
@property
- def is_delete(self):
+ def is_delete(self): # type: ignore[override]
return self._proxied.is_delete
@property
- def is_dml(self):
+ def is_dml(self): # type: ignore[override]
return self._proxied.is_dml
def spoil(self) -> NullLambdaStatement:
table.dispatch.before_parent_attach(table, metadata)
metadata._add_table(name, schema, table)
try:
- table.__init__(name, metadata, *args, _no_init=False, **kw)
+ table.__init__(name, metadata, *args, _no_init=False, **kw) # type: ignore[misc] # noqa: E501
table.dispatch.after_parent_attach(table, metadata)
return table
except Exception:
return _DefaultDescriptionTuple._from_column_default(self.onupdate)
@util.memoized_property
- def _gen_static_annotations_cache_key(self) -> bool: # type: ignore
+ def _gen_static_annotations_cache_key(self) -> bool:
"""special attribute used by cache key gen, if true, we will
use a static cache key for the annotations dictionary, else we
will generate a new cache key for annotations each time.
)
@property
- def sort_key_function(self):
+ def sort_key_function(self): # type: ignore[override]
if self._sort_key_function is NO_ARG:
return self._db_value_for_elem
else:
return self._sort_key_function
@property
- def native(self):
+ def native(self): # type: ignore[override]
return self.native_enum
def _db_value_for_elem(self, elem):
comparator_factory = Comparator
- @property # type: ignore # mypy property bug
+ @property
def should_evaluate_none(self):
"""Alias of :attr:`_types.JSON.none_as_null`"""
return not self.none_as_null
return _python_UUID if self.as_uuid else str
@property
- def native(self):
+ def native(self): # type: ignore[override]
return self.native_uuid
def coerce_compared_value(self, op, value):
def _is_compiled() -> bool:
"""Utility function to indicate if this module is compiled or not."""
- return cython.compiled # type: ignore[no-any-return]
+ return cython.compiled # type: ignore[no-any-return,unused-ignore]
# END GENERATED CYTHON IMPORT
def _is_compiled() -> bool:
"""Utility function to indicate if this module is compiled or not."""
- return cython.compiled # type: ignore[no-any-return]
+ return cython.compiled # type: ignore[no-any-return,unused-ignore]
# END GENERATED CYTHON IMPORT
\r
\r
class ParentProtocol(Protocol):\r
- name: Mapped[str]\r
+ # Read-only for simplicity, mutable protocol members are complicated,\r
+ # see https://mypy.readthedocs.io/en/latest/common_issues.html#covariant-subtyping-of-mutable-protocol-members-is-rejected\r
+ @property\r
+ def name(self) -> Mapped[str]: ...\r
\r
\r
class ChildProtocol(Protocol):\r
def _is_compiled() -> bool:
"""Utility function to indicate if this module is compiled or not."""
- return cython.compiled # type: ignore[no-any-return]
+ return cython.compiled # type: ignore[no-any-return,unused-ignore]
# END GENERATED CYTHON IMPORT\
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
