# See https://pre-commit.com/hooks.html for more hooks
repos:
- repo: https://github.com/python/black/
- rev: 19.3b0
+ rev: 19.10b0
hooks:
- id: black
- args: [-l 79]
- repo: https://github.com/sqlalchemyorg/zimports/
rev: master
from .engine import engine_from_config # noqa nosort
-__version__ = '1.3.14'
+__version__ = "1.3.14"
def __go(lcls):
return tuple(int(x) for x in connection.connection.version.split("."))
def is_disconnect(self, e, connection, cursor):
- error, = e.args
+ (error,) = e.args
if isinstance(
e, (self.dbapi.InterfaceError, self.dbapi.DatabaseError)
) and "not connected" in str(e):
util.warn("Could not parse CHECK constraint text: %r" % src)
sqltext = ""
else:
- sqltext = re.sub(r'^\((.+)\)$', r'\1', m.group(1))
+ sqltext = re.sub(r"^\((.+)\)$", r"\1", m.group(1))
entry = {"name": name, "sqltext": sqltext}
if m and m.group(2):
entry["dialect_options"] = {"not_valid": True}
self._orig_processors = None
if context.result_column_struct:
- result_columns, cols_are_ordered, textual_ordered = (
- context.result_column_struct
- )
+ (
+ result_columns,
+ cols_are_ordered,
+ textual_ordered,
+ ) = context.result_column_struct
num_ctx_cols = len(result_columns)
else:
result_columns = (
--- /dev/null
+[tool.black]
+line-length = 79
RETURN x > 0;
END;$BODY$;
"""
- sa.event.listen(meta, "before_create",
- sa.DDL(udf_create))
- sa.event.listen(meta, "after_drop",
- sa.DDL("DROP FUNCTION is_positive(integer)"))
+ sa.event.listen(meta, "before_create", sa.DDL(udf_create))
+ sa.event.listen(
+ meta, "after_drop", sa.DDL("DROP FUNCTION is_positive(integer)")
+ )
Table(
"pgsql_cc",
def test_simple_lazy_clause_no_race_on_generate(self):
User, Address = self._o2m_fixture()
- expr1, paramdict1 = (
- User.addresses.property._lazy_strategy._simple_lazy_clause
- )
+ (
+ expr1,
+ paramdict1,
+ ) = User.addresses.property._lazy_strategy._simple_lazy_clause
# delete the attr, as though a concurrent thread is also generating it
del User.addresses.property._lazy_strategy._simple_lazy_clause
- expr2, paramdict2 = (
- User.addresses.property._lazy_strategy._simple_lazy_clause
- )
+ (
+ expr2,
+ paramdict2,
+ ) = User.addresses.property._lazy_strategy._simple_lazy_clause
eq_(paramdict1, paramdict2)
mapper(Order, o)
def test_bulk_save_return_defaults(self):
- User, = self.classes("User")
+ (User,) = self.classes("User")
s = Session()
objects = [User(name="u1"), User(name="u2"), User(name="u3")]
eq_(objects[0].__dict__["id"], 1)
def test_bulk_save_mappings_preserve_order(self):
- User, = self.classes("User")
+ (User,) = self.classes("User")
s = Session()
)
def test_bulk_save_no_defaults(self):
- User, = self.classes("User")
+ (User,) = self.classes("User")
s = Session()
objects = [User(name="u1"), User(name="u2"), User(name="u3")]
assert "id" not in objects[0].__dict__
def test_bulk_save_updated_include_unchanged(self):
- User, = self.classes("User")
+ (User,) = self.classes("User")
s = Session(expire_on_commit=False)
objects = [User(name="u1"), User(name="u2"), User(name="u3")]
)
def test_bulk_update(self):
- User, = self.classes("User")
+ (User,) = self.classes("User")
s = Session(expire_on_commit=False)
objects = [User(name="u1"), User(name="u2"), User(name="u3")]
)
def test_bulk_insert(self):
- User, = self.classes("User")
+ (User,) = self.classes("User")
s = Session()
with self.sql_execution_asserter() as asserter:
)
def test_bulk_insert_render_nulls(self):
- Order, = self.classes("Order")
+ (Order,) = self.classes("Order")
s = Session()
with self.sql_execution_asserter() as asserter:
def test_update_defaults_refresh_flush_event_no_postupdate(self):
# run the same test as test_update_defaults_refresh_flush_event
# but don't actually use any postupdate functionality
- A, = self.classes("A")
+ (A,) = self.classes("A")
canary = mock.Mock()
event.listen(A, "refresh_flush", canary.refresh_flush)
# run the same test as
# test_update_defaults_dont_expire_on_delete_no_postupdate
# but don't actually use any postupdate functionality
- A, = self.classes("A")
+ (A,) = self.classes("A")
canary = mock.Mock()
event.listen(A, "refresh_flush", canary.refresh_flush)
self._fixture(False)
City = self.classes.City
sess = Session(testing.db)
- c2, = sess.query(City).order_by(City.id).all()
+ (c2,) = sess.query(City).order_by(City.id).all()
eq_([p.id for p in c2.people], [])
self._test_fk_threelevel(False)
def _test_pk(self, passive_updates):
- Engineer, = self.classes("Engineer")
+ (Engineer,) = self.classes("Engineer")
self._mapping_fixture(False, passive_updates)
sess = sa.orm.sessionmaker()()
)
def _test_pk_threelevel(self, passive_updates):
- Owner, = self.classes("Owner")
+ (Owner,) = self.classes("Owner")
self._mapping_fixture(True, passive_updates)
)
def test_determine_local_remote_pairs_o2m_composite_selfref_func_warning(
- self
+ self,
):
self._assert_non_simple_warning(
self._join_fixture_o2m_composite_selfref_func
)
def test_determine_local_remote_pairs_o2m_composite_selfref_func_annotated(
- self
+ self,
):
joincond = self._join_fixture_o2m_composite_selfref_func_annotated()
eq_(
s.commit()
def test_load(self):
- EmployerUser, = self.classes("EmployerUser")
+ (EmployerUser,) = self.classes("EmployerUser")
s = Session()
q = s.query(EmployerUser)