dbapi_con = connection.connection.dbapi_connection
version: Tuple[Union[int, str], ...] = ()
r = re.compile(r"[.\-]")
- for n in r.split(dbapi_con.getinfo(self.dbapi.SQL_DBMS_VER)): # type: ignore[union-attr] # noqa E501
+ for n in r.split(dbapi_con.getinfo(self.dbapi.SQL_DBMS_VER)): # type: ignore[union-attr] # noqa: E501
try:
version += (int(n),)
except ValueError:
def get_isolation_level_values(
self, dbapi_connection: interfaces.DBAPIConnection
) -> List[_IsolationLevel]:
- return super().get_isolation_level_values(dbapi_connection) + [ # type: ignore # noqa E501
+ return super().get_isolation_level_values(dbapi_connection) + [ # type: ignore # noqa: E501
"AUTOCOMMIT"
]
try:
with db.begin() as conn:
# run magic command to get rid of identity sequences
- # https://floo.bar/2019/11/29/drop-the-underlying-sequence-of-an-identity-column/ # noqa E501
+ # https://floo.bar/2019/11/29/drop-the-underlying-sequence-of-an-identity-column/ # noqa: E501
conn.exec_driver_sql("purge recyclebin")
except exc.DatabaseError as err:
log.warning("purge recyclebin command failed: %s", err)
)
-""" # noqa E501
+""" # noqa: E501
from collections import defaultdict
import datetime as dt
:meth:`_engine.Engine.get_execution_options`
- """ # noqa E501
+ """ # noqa: E501
return self._option_cls(self, opt)
def get_execution_options(self) -> _ExecuteOptions:
:ref:`tutorial_update_delete_rowcount` - in the :ref:`unified_tutorial`
- """ # noqa E501
+ """ # noqa: E501
try:
return self.context.rowcount
)
for made_row in made_rows
]
- if sig_row not in uniques and not uniques.add(sig_row) # type: ignore # noqa E501
+ if sig_row not in uniques and not uniques.add(sig_row) # type: ignore # noqa: E501
]
else:
interim_rows = made_rows
real_result = self._real_result if self._real_result else self
if (
- real_result._source_supports_scalars # type: ignore[attr-defined] # noqa E501
+ real_result._source_supports_scalars # type: ignore[attr-defined] # noqa: E501
and len(indexes) == 1
):
self._generate_rows = False
assert dispatch_target_cls is not None
if (
hasattr(dispatch_target_cls, "__slots__")
- and "_slots_dispatch" in dispatch_target_cls.__slots__ # type: ignore # noqa E501
+ and "_slots_dispatch" in dispatch_target_cls.__slots__ # type: ignore # noqa: E501
):
dispatch_target_cls.dispatch = slots_dispatcher(cls)
else:
def leg(fn: Callable[..., Any]) -> Callable[..., Any]:
if not hasattr(fn, "_legacy_signatures"):
fn._legacy_signatures = [] # type: ignore[attr-defined]
- fn._legacy_signatures.append((since, argnames, converter)) # type: ignore[attr-defined] # noqa E501
+ fn._legacy_signatures.append((since, argnames, converter)) # type: ignore[attr-defined] # noqa: E501
return fn
return leg
**{self.value_attr: other}
)
- def __eq__(self, obj: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa E501
+ def __eq__(self, obj: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
# note the has() here will fail for collections; eq_()
# is only allowed with a scalar.
if obj is None:
else:
return self._comparator.has(**{self.value_attr: obj})
- def __ne__(self, obj: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa E501
+ def __ne__(self, obj: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
# note the has() here will fail for collections; eq_()
# is only allowed with a scalar.
return self._comparator.has(
_target_is_object: bool = False
_is_canonical = True
- def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa E501
+ def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
# special case "is None" to check for no related row as well
expr = self._criterion_exists(
self.remote_attr.operate(operators.eq, other)
the current scope. A function such as ``asyncio.current_task``
may be useful here.
- """ # noqa E501
+ """ # noqa: E501
self.session_factory = session_factory
self.registry = ScopedRegistry(session_factory, scopefunc)
blocking-style code, which will be translated to implicitly async calls
at the point of invoking IO on the database drivers.
- """ # noqa E501
+ """ # noqa: E501
return self.sync_session.get_bind(
mapper=mapper, clause=clause, bind=bind, **kw
def class_logger(cls: Type[_IT]) -> Type[_IT]:
logger = logging.getLogger(_qual_logger_name_for_cls(cls))
- cls._should_log_debug = lambda self: logger.isEnabledFor( # type: ignore[assignment] # noqa E501
+ cls._should_log_debug = lambda self: logger.isEnabledFor( # type: ignore[assignment] # noqa: E501
logging.DEBUG
)
- cls._should_log_info = lambda self: logger.isEnabledFor( # type: ignore[assignment] # noqa E501
+ cls._should_log_info = lambda self: logger.isEnabledFor( # type: ignore[assignment] # noqa: E501
logging.INFO
)
cls.logger = logger
def and_(self, *criteria):
...
- def any(self, criterion=None, **kwargs): # noqa A001
+ def any(self, criterion=None, **kwargs): # noqa: A001
...
def has(self, criterion=None, **kwargs):
:ref:`orm_declarative_dataclasses_mixin` - illustrates special forms
for use with Python dataclasses
- """ # noqa E501
+ """ # noqa: E501
if typing.TYPE_CHECKING:
# test.orm.inheritance.test_basic ->
# EagerTargetingTest.test_adapt_stringency
# OptimizedLoadTest.test_column_expression_joined
- # PolymorphicOnNotLocalTest.test_polymorphic_on_column_prop # noqa E501
+ # PolymorphicOnNotLocalTest.test_polymorphic_on_column_prop # noqa: E501
#
adapted_col = adapter.columns[col]
# mypy seems to get super confused assigning functions to
# attributes
- self._invoke_creator = self._should_wrap_creator(creator) # type: ignore # noqa E501
+ self._invoke_creator = self._should_wrap_creator(creator) # type: ignore # noqa: E501
@_creator.deleter
def _creator(self) -> None:
dbapi_connection: Optional[DBAPIConnection]
@property
- def driver_connection(self) -> Optional[Any]: # type: ignore[override] # mypy#4125 # noqa E501
+ def driver_connection(self) -> Optional[Any]: # type: ignore[override] # mypy#4125 # noqa: E501
if self.dbapi_connection is None:
return None
else:
return {}
@util.memoized_property
- def record_info(self) -> Optional[Dict[str, Any]]: # type: ignore[override] # mypy#4125 # noqa E501
+ def record_info(self) -> Optional[Dict[str, Any]]: # type: ignore[override] # mypy#4125 # noqa: E501
return {}
@classmethod
self._is_valid = False
@property
- def record_info(self) -> Optional[Dict[str, Any]]: # type: ignore[override] # mypy#4125 # noqa E501
+ def record_info(self) -> Optional[Dict[str, Any]]: # type: ignore[override] # mypy#4125 # noqa: E501
return self._connection_record.record_info
def cursor(self, *args: Any, **kwargs: Any) -> DBAPICursor:
_connection_record: Optional[_ConnectionRecord]
@property
- def driver_connection(self) -> Optional[Any]: # type: ignore[override] # mypy#4125 # noqa E501
+ def driver_connection(self) -> Optional[Any]: # type: ignore[override] # mypy#4125 # noqa: E501
if self._connection_record is None:
return None
return self._connection_record.driver_connection
return self._connection_record.info
@property
- def record_info(self) -> Optional[Dict[str, Any]]: # type: ignore[override] # mypy#4125 # noqa E501
+ def record_info(self) -> Optional[Dict[str, Any]]: # type: ignore[override] # mypy#4125 # noqa: E501
if self._connection_record is None:
return None
else:
# will associate with engine.pool
event.listen(engine, 'checkout', my_on_checkout)
- """ # noqa E501
+ """ # noqa: E501
_target_class_doc = "SomeEngineOrPool"
_dispatch_target = Pool
:meth:`_sql.SelectBase.exists` - method to transform a ``SELECT`` to an
``EXISTS`` clause.
- """ # noqa E501
+ """ # noqa: E501
return Exists(__argument)
:ref:`orm_queryguide_execution_options` - documentation on all
ORM-specific execution options
- """ # noqa E501
+ """ # noqa: E501
if "isolation_level" in kw:
raise exc.ArgumentError(
"'isolation_level' execution option may only be specified "
_column_as_key = functools.partial( # type: ignore
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 # noqa: E501
return _column_as_key, _getattr_col_key, _col_bind_name
:ref:`tutorial_insert_returning` - in the :ref:`unified_tutorial`
- """ # noqa E501
+ """ # noqa: E501
if self._return_defaults:
raise exc.InvalidRequestError(
"return_defaults() is already configured on this statement"
:ref:`queryguide_inspection` - ORM background
- """ # noqa E501
+ """ # noqa: E501
meth = DMLState.get_plugin_class(
self
).get_returning_column_descriptions
if lcc > 1:
# multiple elements. Return regular BooleanClauseList
# which will link elements against the operator.
- return cls._construct_raw(operator, convert_clauses) # type: ignore # noqa E501
+ return cls._construct_raw(operator, convert_clauses) # type: ignore # noqa: E501
elif lcc == 1:
# just one element. return it as a single boolean element,
# not a list and discard the operator.
- return convert_clauses[0] # type: ignore[no-any-return] # noqa E501
+ return convert_clauses[0] # type: ignore[no-any-return] # noqa: E501
else:
# no elements period. deprecated use case. return an empty
# ClauseList construct that generates nothing unless it has
},
version="1.4",
)
- return cls._construct_raw(operator) # type: ignore[no-any-return] # noqa E501
+ return cls._construct_raw(operator) # type: ignore[no-any-return] # noqa: E501
@classmethod
def _construct_for_whereclause(
:meth:`_functions.FunctionElement.column_valued`
- """ # noqa E501
+ """ # noqa: E501
return ScalarFunctionColumn(self, name, type_)
:meth:`_sql.TableValuedAlias.render_derived` - renders the alias
using a derived column clause, e.g. ``AS name(col1, col2, ...)``
- """ # noqa 501
+ """ # noqa: 501
new_func = self._generate()
:meth:`_functions.FunctionElement.table_valued`
- """ # noqa 501
+ """ # noqa: 501
return self.alias(name=name).column
:meth:`_functions.FunctionElement.table_valued` - generates table-valued
SQL function expressions.
- """ # noqa E501
+ """ # noqa: E501
return self.c
@util.ro_memoized_property
inherit_cache = True
-class max(ReturnTypeFromArgs[_T]): # noqa A001
+class max(ReturnTypeFromArgs[_T]): # noqa: A001
"""The SQL MAX() aggregate function."""
inherit_cache = True
-class min(ReturnTypeFromArgs[_T]): # noqa A001
+class min(ReturnTypeFromArgs[_T]): # noqa: A001
"""The SQL MIN() aggregate function."""
inherit_cache = True
-class sum(ReturnTypeFromArgs[_T]): # noqa A001
+class sum(ReturnTypeFromArgs[_T]): # noqa: A001
"""The SQL SUM() aggregate function."""
inherit_cache = True
if hasattr(left, "__sa_operate__"):
return left.operate(self, right, *other, **kwargs)
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 # noqa: E501
else:
raise exc.InvalidRequestError(
f"Custom operator {self.opstring!r} can't be used with "
by the SQLAlchemy dialect.
.. note:: setting this flag to ``False`` will not provide
- case-insensitive behavior for table reflection; table reflection
- will always search for a mixed-case name in a case sensitive
- fashion. Case insensitive names are specified in SQLAlchemy only
- by stating the name with all lower case characters.
+ case-insensitive behavior for table reflection; table reflection
+ will always search for a mixed-case name in a case sensitive
+ fashion. Case insensitive names are specified in SQLAlchemy only
+ by stating the name with all lower case characters.
:param quote_schema: same as 'quote' but applies to the schema identifier.
specify the special symbol :attr:`.BLANK_SCHEMA`.
.. versionadded:: 1.0.14 Added the :attr:`.BLANK_SCHEMA` symbol to
- allow a :class:`_schema.Table`
- to have a blank schema name even when the
- parent :class:`_schema.MetaData` specifies
- :paramref:`_schema.MetaData.schema`.
+ allow a :class:`_schema.Table`
+ to have a blank schema name even when the
+ parent :class:`_schema.MetaData` specifies
+ :paramref:`_schema.MetaData.schema`.
The quoting rules for the schema name are the same as those for the
``name`` parameter, in that quoting is applied for reserved words or
See the documentation regarding an individual dialect at
:ref:`dialect_toplevel` for detail on documented arguments.
- """ # noqa E501
+ """ # noqa: E501
# __init__ is overridden to prevent __new__ from
# calling the superclass constructor.
parameter to :class:`_schema.Column`.
- """ # noqa E501
+ """ # noqa: E501, RST201, RST202
name = kwargs.pop("name", None)
type_ = kwargs.pop("type_", None)
object, returning a copy of this :class:`_expression.FromClause`.
"""
- return util.preloaded.sql_util.ClauseAdapter(alias).traverse( # type: ignore # noqa E501
+ return util.preloaded.sql_util.ClauseAdapter(alias).traverse( # type: ignore # noqa: E501
self
)
.. versionadded:: 1.4
- """ # noqa E501
+ """ # noqa: E501
LABEL_STYLE_TABLENAME_PLUS_COL = 1
"""Label style indicating all columns should be labeled as
:ref:`tutorial_functions_table_valued` - in the :ref:`unified_tutorial`
- """ # noqa E501
+ """ # noqa: E501
__visit_name__ = "table_valued_alias"
datatype specification with each column. This is a special syntax
currently known to be required by PostgreSQL for some SQL functions.
- """ # noqa E501
+ """ # noqa: E501
# note: don't use the @_generative system here, keep a reference
# to the original object. otherwise you can have re-use of the
.. versionadded:: 1.4.23
- """ # noqa E501
+ """ # noqa: E501
# memoizations should be cleared here as of
# I95c560ffcbfa30b26644999412fb6a385125f663 , asserting this
"_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 # 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 # noqa: E501
return shallow_to_dict(self)
def _shallow_copy_to(
op_fn, addtl_kw = default_comparator.operator_lookup[op.__name__]
if kwargs:
addtl_kw = addtl_kw.union(kwargs)
- return op_fn(self.expr, op, other, reverse=True, **addtl_kw) # type: ignore # noqa E501
+ return op_fn(self.expr, op, other, reverse=True, **addtl_kw) # type: ignore # noqa: E501
def _adapt_expression(
self,
def comparator_factory( # type: ignore # 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 # noqa: E501
return self.impl.comparator_factory
else:
# reconcile the Comparator class on the impl with that
# of TypeDecorator
return type(
"TDComparator",
- (TypeDecorator.Comparator, self.impl.comparator_factory), # type: ignore # noqa E501
+ (TypeDecorator.Comparator, self.impl.comparator_factory), # type: ignore # noqa: E501
{},
)
# 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 # noqa: E501
return self.impl_instance.sort_key_function
def __repr__(self) -> str:
from .base import ColumnSet
from .cache_key import HasCacheKey # noqa
from .ddl import sort_tables # noqa
-from .elements import _find_columns # noqa
+from .elements import _find_columns
from .elements import _label_reference
from .elements import _textual_label_reference
from .elements import BindParameter
-from .elements import ClauseElement # noqa
+from .elements import ClauseElement
from .elements import ColumnClause
from .elements import ColumnElement
from .elements import Grouping
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 # noqa: E501
else:
- return meth(self, **kw) # type: ignore # noqa E501
+ return meth(self, **kw) # type: ignore # noqa: E501
cls._compiler_dispatch = ( # type: ignore
cls._original_compiler_dispatch
def greenlet(self):
def go(config):
try:
- import greenlet # noqa F401
+ import greenlet # noqa: F401
except ImportError:
return False
else:
self.data = data
self._unique = {}
if via:
- self._data_appender = getattr(data, via) # type: ignore[assignment] # noqa E501
+ self._data_appender = getattr(data, via) # type: ignore[assignment] # noqa: E501
elif hasattr(data, "append"):
- self._data_appender = cast("List[_T]", data).append # type: ignore[assignment] # noqa E501
+ self._data_appender = cast("List[_T]", data).append # type: ignore[assignment] # noqa: E501
elif hasattr(data, "add"):
- self._data_appender = cast("Set[_T]", data).add # type: ignore[assignment] # noqa E501
+ self._data_appender = cast("Set[_T]", data).add # type: ignore[assignment] # noqa: E501
def append(self, item: _T) -> None:
id_ = id(item)
"loop is already running; can't call await_() here. "
"Was IO attempted in an unexpected place?"
)
- return loop.run_until_complete(awaitable) # type: ignore[no-any-return] # noqa E501
+ return loop.run_until_complete(awaitable) # type: ignore[no-any-return] # noqa: E501
return current.driver.switch(awaitable) # type: ignore[no-any-return]
return self # type: ignore
def union(self, *other: Iterable[_S]) -> "OrderedSet[Union[_T, _S]]":
- result: "OrderedSet[Union[_T, _S]]" = self.__class__(self) # type: ignore # noqa E501
+ result: "OrderedSet[Union[_T, _S]]" = self.__class__(self) # type: ignore # noqa: E501
for o in other:
result.update(o)
return result
have_greenlet = False
greenlet_error = None
try:
- import greenlet # type: ignore # noqa F401
+ import greenlet # type: ignore # noqa: F401
except ImportError as e:
greenlet_error = str(e)
pass
from ._concurrency_py3k import AsyncAdaptedLock as AsyncAdaptedLock
from ._concurrency_py3k import (
_util_async_run as _util_async_run,
- ) # noqa F401
+ ) # noqa: F401
from ._concurrency_py3k import (
- _util_async_run_coroutine_function as _util_async_run_coroutine_function, # noqa F401, E501
+ _util_async_run_coroutine_function as _util_async_run_coroutine_function, # noqa: F401, E501
)
if not typing.TYPE_CHECKING and not have_greenlet:
else ""
)
- def is_exit_exception(e): # noqa F811
+ def is_exit_exception(e):
return not isinstance(e, Exception)
- def await_only(thing): # type: ignore # noqa F811
+ def await_only(thing): # type: ignore
_not_implemented()
- def await_fallback(thing): # type: ignore # noqa F81
+ def await_fallback(thing): # type: ignore
return thing
- def greenlet_spawn(fn, *args, **kw): # type: ignore # noqa F81
+ def greenlet_spawn(fn, *args, **kw): # type: ignore
_not_implemented()
- def AsyncAdaptedLock(*args, **kw): # type: ignore # noqa F81
+ def AsyncAdaptedLock(*args, **kw): # type: ignore
_not_implemented()
- def _util_async_run(fn, *arg, **kw): # type: ignore # noqa F81
+ def _util_async_run(fn, *arg, **kw): # type: ignore
return fn(*arg, **kw)
- def _util_async_run_coroutine_function(fn, *arg, **kw): # type: ignore # noqa F81
+ def _util_async_run_coroutine_function(fn, *arg, **kw): # type: ignore
_not_implemented()
fn.__init__, no_self=no_self, _is_init=True
)
elif hasattr(fn, "__func__"):
- return compat.inspect_getfullargspec(fn.__func__) # type: ignore[attr-defined] # noqa E501
+ return compat.inspect_getfullargspec(fn.__func__) # type: ignore[attr-defined] # noqa: E501
elif hasattr(fn, "__call__"):
if inspect.ismethod(fn.__call__): # type: ignore [operator]
return get_callable_argspec(
else:
code = (
"def %(name)s%(grouped_args)s:\n"
- " return %(self_arg)s._proxied.%(name)s(%(apply_kw_proxied)s)" # noqa E501
+ " return %(self_arg)s._proxied.%(name)s(%(apply_kw_proxied)s)" # noqa: E501
% metadata
)
if isinstance(argtype, tuple):
raise exc.ArgumentError(
"Argument '%s' is expected to be one of type %s, got '%s'"
- % (name, " or ".join("'%s'" % a for a in argtype), type(arg)) # type: ignore # noqa E501
+ % (name, " or ".join("'%s'" % a for a in argtype), type(arg)) # type: ignore # noqa: E501
)
else:
raise exc.ArgumentError(
if typing.TYPE_CHECKING or compat.py310:
from typing import Annotated as Annotated
else:
- from typing_extensions import Annotated as Annotated # noqa F401
+ from typing_extensions import Annotated as Annotated # noqa: F401
if typing.TYPE_CHECKING or compat.py38:
from typing import Literal as Literal
from typing import Protocol as Protocol
from typing import TypedDict as TypedDict
else:
- from typing_extensions import Literal as Literal # noqa F401
- from typing_extensions import Protocol as Protocol # noqa F401
- from typing_extensions import TypedDict as TypedDict # noqa F401
+ from typing_extensions import Literal as Literal # noqa: F401
+ from typing_extensions import Protocol as Protocol # noqa: F401
+ from typing_extensions import TypedDict as TypedDict # noqa: F401
# copied from TypeShed, required in order to implement
# MutableMapping.update()
from typing_extensions import Concatenate as Concatenate
from typing_extensions import ParamSpec as ParamSpec
else:
- from typing import Concatenate as Concatenate # noqa F401
- from typing import ParamSpec as ParamSpec # noqa F401
+ from typing import Concatenate as Concatenate # noqa: F401
+ from typing import ParamSpec as ParamSpec # noqa: F401
def de_stringify_annotation(
for i in range(100):
# test counts assume objects remain in the session
# from previous run
- r = q.all() # noqa F841
+ r = q.all() # noqa: F841
go()
for i in range(100):
# test counts assume objects remain in the session
# from previous run
- r = q.all() # noqa F841
+ r = q.all() # noqa: F841
go()
for i in range(100):
# test counts assume objects remain in the session
# from previous run
- r = q.all() # noqa F841
+ r = q.all() # noqa: F841
go()
for i in range(100):
# test counts assume objects remain in the session
# from previous run
- r = q.all() # noqa F841
+ r = q.all() # noqa: F841
go()
for i in range(100):
# test counts assume objects remain in the session
# from previous run
- r = q.all() # noqa F841
+ r = q.all() # noqa: F841
go()
for i in range(100):
# test counts assume objects remain in the session
# from previous run
- r = q.all() # noqa F841
+ r = q.all() # noqa: F841
go()
for i in range(100):
# test counts assume objects remain in the session
# from previous run
- r = q.all() # noqa F841
+ r = q.all() # noqa: F841
go()
for i in range(100):
# test counts assume objects remain in the session
# from previous run
- r = q.all() # noqa F841
+ r = q.all() # noqa: F841
go()
for i in range(100):
# test counts assume objects remain in the session
# from previous run
- r = q.all() # noqa F841
+ r = q.all() # noqa: F841
go()
oracle.FLOAT(16),
), # using conversion
(FLOAT(), DOUBLE_PRECISION()),
- # from https://docs.oracle.com/cd/B14117_01/server.101/b10758/sqlqr06.htm # noqa E501
+ # from https://docs.oracle.com/cd/B14117_01/server.101/b10758/sqlqr06.htm # noqa: E501
# DOUBLE PRECISION == precision 126
# REAL == precision 63
(oracle.FLOAT(126), DOUBLE_PRECISION()),
u1 = User()
-# EXPECTED_MYPY: Incompatible types in assignment (expression has type "Optional[str]", variable has type "str") # noqa E501
+# EXPECTED_MYPY: Incompatible types in assignment (expression has type "Optional[str]", variable has type "str") # noqa: E501
p: str = u1.name
u1 = User()
-# EXPECTED_MYPY: Incompatible types in assignment (expression has type "Optional[str]", variable has type "Optional[int]") # noqa E501
+# EXPECTED_MYPY: Incompatible types in assignment (expression has type "Optional[str]", variable has type "Optional[int]") # noqa: E501
p: Optional[int] = u1.name
u1 = User()
-# EXPECTED_MYPY: Incompatible types in assignment (expression has type "Optional[str]", variable has type "str") # noqa E501
+# EXPECTED_MYPY: Incompatible types in assignment (expression has type "Optional[str]", variable has type "str") # noqa: E501
p: str = u1.name
__tablename__ = "user"
id = Column(Integer(), primary_key=True)
- # EXPECTED: Left hand assignment 'name: "int"' not compatible with ORM mapped expression # noqa E501
+ # EXPECTED: Left hand assignment 'name: "int"' not compatible with ORM mapped expression # noqa: E501
name: int = Column(String())
__tablename__ = "address"
id = Column(Integer, primary_key=True)
- # EXPECTED: Can't infer type from ORM mapped expression assigned to attribute 'user_id'; # noqa E501
+ # EXPECTED: Can't infer type from ORM mapped expression assigned to attribute 'user_id'; # noqa: E501
user_id = Column(ForeignKey("user.id"))
email_address = Column(String)
is_re = bool(m.group(2))
is_type = bool(m.group(3))
- expected_msg = re.sub(r"# noqa ?.*", "", m.group(4))
+ expected_msg = re.sub(r"# noqa[:]? ?.*", "", m.group(4))
if is_type:
is_mypy = is_re = True
expected_msg = f'Revealed type is "{expected_msg}"'
for idx, (typ, errmsg) in enumerate(output):
if is_re:
if re.match(
- rf".*{filename}\:{num}\: {typ}\: {prefix}{msg}", # noqa E501
+ rf".*{filename}\:{num}\: {typ}\: {prefix}{msg}", # noqa: E501
errmsg,
):
break
__sa_dataclass_metadata_key__ = "sa"
id: Mapped[int] = mapped_column(primary_key=True)
- bs: List["B"] = dataclasses.field( # noqa: F821
+ bs: List["B"] = dataclasses.field(
default_factory=list, metadata={"sa": relationship()}
)
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[str] = mapped_column()
- bs: Mapped[List["B"]] = relationship( # noqa F821
- back_populates="a"
- )
+ bs: Mapped[List["B"]] = relationship(back_populates="a")
class B(decl_base):
__tablename__ = "b"
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[str] = mapped_column()
- bs_list: Mapped[List["B"]] = relationship( # noqa F821
- viewonly=True
- )
- bs_set: Mapped[Set["B"]] = relationship(viewonly=True) # noqa F821
- bs_list_warg: Mapped[List["B"]] = relationship( # noqa F821
- "B", viewonly=True
- )
- bs_set_warg: Mapped[Set["B"]] = relationship( # noqa F821
- "B", viewonly=True
- )
+ bs_list: Mapped[List["B"]] = relationship(viewonly=True)
+ bs_set: Mapped[Set["B"]] = relationship(viewonly=True)
+ bs_list_warg: Mapped[List["B"]] = relationship("B", viewonly=True)
+ bs_set_warg: Mapped[Set["B"]] = relationship("B", viewonly=True)
class B(decl_base):
__tablename__ = "b"
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[str] = mapped_column()
- bs: Mapped[Dict[str, "B"]] = relationship() # noqa F821
+ bs: Mapped[Dict[str, "B"]] = relationship()
class B(decl_base):
__tablename__ = "b"
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[str] = mapped_column()
- bs: Mapped[MappedCollection[str, "B"]] = relationship( # noqa F821
+ bs: Mapped[MappedCollection[str, "B"]] = relationship(
collection_class=attribute_mapped_collection("name")
)
name: Mapped[str50]
- employees: Mapped[Set["Person"]] = relationship() # noqa F821
+ employees: Mapped[Set["Person"]] = relationship()
class Person(decl_base):
__tablename__ = "person"
with testing.expect_deprecated(
"The mapper.non_primary parameter is deprecated"
):
- m = self.mapper_registry.map_imperatively( # noqa F841
+ m = self.mapper_registry.map_imperatively( # noqa: F841
User,
users,
non_primary=True,
with testing.expect_deprecated(
"The mapper.non_primary parameter is deprecated"
):
- m = registry.map_imperatively( # noqa F841
+ m = registry.map_imperatively( # noqa: F841
User,
users,
non_primary=True,
`WITH ORDINALITY AS unnested(unnested, ordinality) ON true
LEFT OUTER JOIN b ON unnested.unnested = b.ref
- """ # noqa 501
+ """ # noqa: 501
a = table("a", column("id"), column("refs"))
b = table("b", column("id"), column("ref"))
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
