import itertools
from sqlalchemy import util
-
from sqlalchemy.sql import visitors
from sqlalchemy.sql.expression import Select
def find_unmatching_froms(query, start_with=None):
- # type: (Select, Optional[FromClause]) -> Tuple[Set[FromClause], FromClause]
- # TODO: It would be nicer to use OrderedSet, but it seems to not be too much optimize, so let's skip for now
+ # type: (Select, Optional[FromClause]) -> Tuple[Set[FromClause], FromClause] # noqa
+ # TODO: It would be nicer to use OrderedSet, but it seems to not
+ # be too much optimized, so let's skip for now
froms = set(query.froms)
if not froms:
return None, None
# type: (Select) -> None
froms, start_with = find_unmatching_froms(query)
if froms:
- template = '''Query\n{query}\nhas FROM elements:\n{froms}\nthat are not joined up to FROM element\n{start}'''
+ template = 'Query\n{query}\nhas FROM elements:\n{froms}\nthat ' \
+ 'are not joined up to FROM element\n{start}'''
indent = ' '
froms_str = '\n'.join('* {elem}'.format(elem=from_) for from_ in froms)
message = template.format(
-from sqlalchemy import select, Integer, event, testing
+from sqlalchemy import event, Integer, select, testing
from sqlalchemy.ext import linter
from sqlalchemy.ext.linter import find_unmatching_froms
-from sqlalchemy.testing import fixtures, expect_warnings
-from sqlalchemy.testing.schema import Table, Column
+from sqlalchemy.testing import expect_warnings, fixtures
from sqlalchemy.testing.mock import patch
+from sqlalchemy.testing.schema import Column, Table
class TestFindUnmatchingFroms(fixtures.TablesTest):
assert not froms
def test_disconnected_subquery(self):
- subq = select([self.a]).where(self.a.c.col_a == self.b.c.col_b).subquery()
+ subq = (
+ select([self.a])
+ .where(self.a.c.col_a == self.b.c.col_b)
+ .subquery()
+ )
stmt = select([self.c]).select_from(subq)
froms, start = find_unmatching_froms(stmt, self.c)
assert froms == {self.c}
def test_now_connect_it(self):
- subq = select([self.a]).where(self.a.c.col_a == self.b.c.col_b).subquery()
- stmt = select([self.c]).select_from(subq).where(self.c.c.col_c == subq.c.col_a)
+ subq = (
+ select([self.a])
+ .where(self.a.c.col_a == self.b.c.col_b)
+ .subquery()
+ )
+ stmt = (
+ select([self.c])
+ .select_from(subq)
+ .where(self.c.c.col_c == subq.c.col_a)
+ )
froms, start = find_unmatching_froms(stmt)
assert not froms
query = (
select([self.a])
.select_from(
- self.a.join(self.b.join(self.c, self.b.c.col_b == self.c.c.col_c), self.a.c.col_a == self.b.c.col_b)
+ self.a.join(
+ self.b.join(self.c, self.b.c.col_b == self.c.c.col_c),
+ self.a.c.col_a == self.b.c.col_b,
+ )
)
)
froms, start = find_unmatching_froms(query)