# See https://pre-commit.com/hooks.html for more hooks
repos:
- repo: https://github.com/python/black
- rev: 20.8b1
+ rev: 21.5b1
hooks:
- id: black
- repo: https://github.com/sqlalchemyorg/zimports
- rev: 0.3.0
+ rev: v0.4.0
hooks:
- id: zimports
+ args:
+ - --keep-unused-type-checking
- repo: https://github.com/pycqa/flake8
- rev: 3.8.4
+ rev: 3.9.2
hooks:
- id: flake8
additional_dependencies:
@Profiler.profile
def test_bulk_save(n):
- """Individual INSERT/COMMIT pairs using the "bulk" API """
+ """Individual INSERT/COMMIT pairs using the "bulk" API"""
for i in range(n):
session = Session(bind=engine)
class ArmyGlyph(EnemyGlyph):
- """Describe an enemy that's part of the "army". """
+ """Describe an enemy that's part of the "army"."""
__mapper_args__ = {"polymorphic_identity": "army"}
@event.listens_for(Query, "before_compile", retval=True)
def before_compile(query):
- """ensure all queries for VersionedStartEnd include criteria """
+ """ensure all queries for VersionedStartEnd include criteria"""
for ent in query.column_descriptions:
entity = ent["entity"]
)
def get_select_precolumns(self, select, **kw):
- """ MS-SQL puts TOP, it's version of LIMIT here """
+ """MS-SQL puts TOP, it's version of LIMIT here"""
s = super(MSSQLCompiler, self).get_select_precolumns(select, **kw)
class TIME(sqltypes.TIME):
- """MySQL TIME type. """
+ """MySQL TIME type."""
__visit_name__ = "TIME"
if compat.TYPE_CHECKING:
from typing import Any
+
from .url import URL
"""
def execute(self, object_, *multiparams, **params):
- """Executes the given construct and returns a """
- """:class:`_engine.CursorResult`."""
+ """Executes the given construct and returns a
+ :class:`_engine.CursorResult`.
+ """
raise NotImplementedError()
def scalar(self, object_, *multiparams, **params):
if util.TYPE_CHECKING:
from typing import Any
- from typing import List
- from typing import Optional
from typing import Int
from typing import Iterator
+ from typing import List
from typing import Mapping
+ from typing import Optional
+
from ...engine.result import Row
from typing import Any
from typing import List
from typing import Optional
+
from .mapper import Mapper
from .util import AliasedInsp
if util.TYPE_CHECKING:
- from .util import AliasedInsp
from typing import Union
+ from .util import AliasedInsp
+
def remote(expr):
"""Annotate a portion of a primaryjoin expression
from ..sql.base import Generative
if util.TYPE_CHECKING:
- from .context import QueryContext
from typing import Sequence
+
+ from .context import QueryContext
from ..sql.elements import ColumnElement
return self
def _ungroup(self):
- """Return this :class:`_expression.ClauseElement` """
- """without any groupings."""
+ """Return this :class:`_expression.ClauseElement`
+ without any groupings.
+ """
return self
class DMLSelectRole(SQLRole):
- """A SELECT statement embedded in DML, typically INSERT from SELECT """
+ """A SELECT statement embedded in DML, typically INSERT from SELECT"""
_role_name = "SELECT statement or equivalent textual object"
@testing.fixture
def literal_round_trip(self, metadata, connection):
- """test literal rendering """
+ """test literal rendering"""
# for literal, we test the literal render in an INSERT
# into a typed column. we can then SELECT it back as its
conn.exec_driver_sql("drop table test_table")
def test_reflection(self):
- """test the resolution of the synonym/dblink. """
+ """test the resolution of the synonym/dblink."""
m = MetaData()
t = Table(
class DialectTest(fixtures.TestBase):
- """python-side dialect tests. """
+ """python-side dialect tests."""
def test_version_parsing(self):
def mock_conn(res):
class SpecialTypesTest(fixtures.TablesTest, ComparesTables):
- """test DDL and reflection of PG-specific types """
+ """test DDL and reflection of PG-specific types"""
__only_on__ = ("postgresql >= 8.3.0",)
__backend__ = True
)
def test_prop_on_base(self):
- """test [ticket:2670] """
+ """test [ticket:2670]"""
counter = mock.Mock()
from .user import HasUser
if TYPE_CHECKING:
- from .user import User # noqa
- from sqlalchemy import Integer, Column # noqa
+ from sqlalchemy import Column # noqa
+ from sqlalchemy import Integer # noqa
from sqlalchemy.orm import RelationshipProperty # noqa
+ from .user import User # noqa
class Address(Base, HasUser):
def step_numbering(step):
- """ order in whole steps """
+ """order in whole steps"""
def f(index, collection):
return step * index
__tablename__ = "person"
class Engineer(Person):
- """ single table inheritance, no extra cols """
+ """single table inheritance, no extra cols"""
class Manager(Person):
- """ single table inheritance, no extra cols """
+ """single table inheritance, no extra cols"""
is_(Engineer.id.property.columns[0], Person.__table__.c.id)
is_(Manager.id.property.columns[0], Person.__table__.c.id)
class JoinedloadWPolyOfTypeContinued(
fixtures.DeclarativeMappedTest, testing.AssertsCompiledSQL
):
- """test for #5082 """
+ """test for #5082"""
@classmethod
def setup_classes(cls):
class ContainsEagerMultipleOfType(
fixtures.DeclarativeMappedTest, testing.AssertsCompiledSQL
):
- """test for #5107 """
+ """test for #5107"""
__dialect__ = "default"
def test_autobegin_attr_change(
self, case_, autocommit, begin, modify_unconditional, expire_on_commit
):
- """test :ticket:`6360` """
+ """test :ticket:`6360`"""
User, users = self.classes.User, self.tables.users
not_in(jane, sess)
def test_update_with_filter_statement(self):
- """test for [ticket:4556] """
+ """test for [ticket:4556]"""
User = self.classes.User
)
def test_recursive_union_alias_two(self):
- """"""
-
# I know, this is the PG VALUES keyword,
# we're cheating here. also yes we need the SELECT,
# sorry PG.
self._assert_seq_result(testing.db.scalar(s.next_value()))
def test_func_implicit_connectionless_scalar(self):
- """test func.next_value().execute()/.scalar() works. """
+ """test func.next_value().execute()/.scalar() works."""
with testing.expect_deprecated_20(
r"The MetaData.bind argument is deprecated"
)
def test_scalar_subquery_wo_type(self):
- """ test for :ticket:`6181` """
+ """test for :ticket:`6181`"""
m = MetaData()
t = Table("t", m, Column("a", Integer))
)
def test_expanding_in_dont_alter_compiled(self, connection):
- """test for issue #5048 """
+ """test for issue #5048"""
class NameWithProcess(TypeDecorator):
impl = String
)
def test_untyped_scalar_subquery(self):
- """test for :ticket:`6181` """
+ """test for :ticket:`6181`"""
c = column("q")
subq = select(c).scalar_subquery()
)
def test_labels_no_collision_index(self):
- """test for [ticket:4911] """
+ """test for [ticket:4911]"""
t = Table(
"foo",
# in case it requires a version pin
pydocstyle
pygments
- black==20.8b1
+ black==21.5b1
commands =
flake8 ./lib/ ./test/ ./examples/ setup.py doc/build/conf.py {posargs}
black --check .