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
:ref:`tutorial_update_delete_rowcount` - in the :ref:`unified_tutorial`
- """ # noqa E501
+ """ # noqa: E501
try:
return self.context.rowcount
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
:ref:`orm_declarative_dataclasses_mixin` - illustrates special forms
for use with Python dataclasses
- """ # noqa E501
+ """ # noqa: E501
def __init__(self, fget, cascading=False):
super(declared_attr, self).__init__(fget)
# 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]
: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
: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 ColumnCollection(
columns=[(col.key, col) for col in self._all_selected_columns]
inherit_cache = True
-class max(ReturnTypeFromArgs): # noqa A001
+class max(ReturnTypeFromArgs): # noqa: A001
"""The SQL MAX() aggregate function."""
inherit_cache = True
-class min(ReturnTypeFromArgs): # noqa A001
+class min(ReturnTypeFromArgs): # noqa: A001
"""The SQL MIN() aggregate function."""
inherit_cache = True
-class sum(ReturnTypeFromArgs): # noqa A001
+class sum(ReturnTypeFromArgs): # noqa: A001
"""The SQL SUM() aggregate function."""
inherit_cache = True
parameter to :class:`_schema.Column`.
- """ # noqa E501
+ """ # noqa: E501, RST201, RST202
name = kwargs.pop("name", None)
type_ = kwargs.pop("type_", None)
.. versionadded:: 1.4
-""", # noqa E501
+""", # noqa: E501
)
LABEL_STYLE_TABLENAME_PLUS_COL = util.symbol(
.. versionadded:: 1.4
-""", # noqa E501
+""", # noqa: E501
)
: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
:meth:`_sql.SelectBase.exists` - method to transform a ``SELECT`` to an
``EXISTS`` clause.
- """ # noqa E501
+ """ # noqa: E501
if args and isinstance(args[0], (SelectBase, ScalarSelect)):
s = args[0]
else:
def greenlet(self):
def go(config):
try:
- import greenlet # noqa F401
+ import greenlet # noqa: F401
except ImportError:
return False
else:
if compat.py3k:
try:
- import greenlet # noqa F401
+ import greenlet # noqa: F401
except ImportError as e:
greenlet_error = str(e)
else:
from ._concurrency_py3k import greenlet_spawn
from ._concurrency_py3k import is_exit_exception
from ._concurrency_py3k import AsyncAdaptedLock
- from ._concurrency_py3k import _util_async_run # noqa F401
+ from ._concurrency_py3k import _util_async_run # noqa: F401
from ._concurrency_py3k import (
_util_async_run_coroutine_function,
- ) # noqa F401, E501
- from ._concurrency_py3k import asyncio # noqa F401
+ ) # noqa: F401, E501
+ from ._concurrency_py3k import asyncio # noqa: F401
# does not need greennlet, just Python 3
- from ._compat_py3k import asynccontextmanager # noqa F401
+ from ._compat_py3k import asynccontextmanager # noqa: F401
if not have_greenlet:
- asyncio = None # noqa F811
+ asyncio = None # noqa: F811
def _not_implemented():
# this conditional is to prevent pylance from considering
else ""
)
- def is_exit_exception(e): # noqa F811
+ def is_exit_exception(e): # noqa: F811
return not isinstance(e, Exception)
- def await_only(thing): # noqa F811
+ def await_only(thing): # noqa: F811
_not_implemented()
- def await_fallback(thing): # noqa F81
+ def await_fallback(thing): # noqa: F811
return thing
- def greenlet_spawn(fn, *args, **kw): # noqa F81
+ def greenlet_spawn(fn, *args, **kw): # noqa: F811
_not_implemented()
- def AsyncAdaptedLock(*args, **kw): # noqa F81
+ def AsyncAdaptedLock(*args, **kw): # noqa: F811
_not_implemented()
- def _util_async_run(fn, *arg, **kw): # noqa F81
+ def _util_async_run(fn, *arg, **kw): # noqa: F811
return fn(*arg, **kw)
- def _util_async_run_coroutine_function(fn, *arg, **kw): # noqa F81
+ def _util_async_run_coroutine_function(fn, *arg, **kw): # noqa: F811
_not_implemented()
else:
code = (
"def %(name)s(%(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
)
def has_compiled_ext():
try:
- from sqlalchemy import cimmutabledict # noqa F401
- from sqlalchemy import cprocessors # noqa F401
- from sqlalchemy import cresultproxy # noqa F401
+ from sqlalchemy import cimmutabledict # noqa: F401
+ from sqlalchemy import cprocessors # noqa: F401
+ from sqlalchemy import cresultproxy # noqa: F401
return True
except ImportError:
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()
eq_(str(err), message)
# unicode accessor decodes to utf-8
- eq_(unicode(err), util.u("some message méil")) # noqa F821
+ eq_(unicode(err), util.u("some message méil")) # noqa: F821
else:
eq_(str(err), util.u("some message méil"))
eq_(str(err), message)
# unicode accessor decodes to utf-8
- eq_(unicode(err), util.u("some message m\\xe9il")) # noqa F821
+ eq_(unicode(err), util.u("some message m\\xe9il")) # noqa: F821
else:
eq_(str(err), util.u("some message m\\xe9il"))
err = tsa.exc.SQLAlchemyError(message)
if util.py2k:
- eq_(unicode(err), util.u("some message méil")) # noqa F821
+ eq_(unicode(err), util.u("some message méil")) # noqa: F821
else:
eq_(str(err), util.u("some message méil"))
eq_(str(err), "foo")
if util.py2k:
- eq_(unicode(err), util.u("fóó")) # noqa F821
+ eq_(unicode(err), util.u("fóó")) # noqa: F821
def test_stmt_exception_str_multi_args(self):
err = tsa.exc.SQLAlchemyError("some message", 206)
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)
if m:
is_mypy = bool(m.group(1))
expected_msg = m.group(2)
- expected_msg = re.sub(r"# noqa ?.*", "", m.group(2))
+ expected_msg = re.sub(r"# noqa[:]? ?.*", "", m.group(2))
expected_errors.append(
(num, is_mypy, expected_msg.strip())
)
mach_alias = machines.select()
# note python 2 does not allow parens here; reformat in py3 only
- with DeprecatedQueryTest._expect_implicit_subquery(), _aliased_join_warning( # noqa E501
+ with DeprecatedQueryTest._expect_implicit_subquery(), _aliased_join_warning( # noqa: E501
"Person->people"
):
self.assert_compile(
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 = mapper( # noqa F841
+ m = mapper( # 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
