--- /dev/null
+.. change::
+ :tags: bug, reflection
+ :tickets: 8100, 8101
+
+ Fixed bugs involving the :paramref:`.Table.include_columns` and the
+ :paramref:`.Table.resolve_fks` parameters on :class:`.Table`; these
+ little-used parameters were apparently not working for columns that refer
+ to foreign key constraints.
+
+ In the first case, not-included columns that refer to foreign keys would
+ still attempt to create a :class:`.ForeignKey` object, producing errors
+ when attempting to resolve the columns for the foreign key constraint
+ within reflection; foreign key constraints that refer to skipped columns
+ are now omitted from the table reflection process in the same way as
+ occurs for :class:`.Index` and :class:`.UniqueConstraint` objects with the
+ same conditions. No warning is produced however, as we likely want to
+ remove the include_columns warnings for all constraints in 2.0.
+
+ In the latter case, the production of table aliases or subqueries would
+ fail on an FK related table not found despite the presence of
+ ``resolve_fks=False``; the logic has been repaired so that if a related
+ table is not found, the :class:`.ForeignKey` object is still proxied to the
+ aliased table or subquery (these :class:`.ForeignKey` objects are normally
+ used in the production of join conditions), but it is sent with a flag that
+ it's not resolvable. The aliased table / subquery will then work normally,
+ with the exception that it cannot be used to generate a join condition
+ automatically, as the foreign key information is missing. This was already
+ the behavior for such foreign key constraints produced using non-reflection
+ methods, such as joining :class:`.Table` objects from different
+ :class:`.MetaData` collections.
information is not transferred.
"""
+
fk = [
- ForeignKey(f.column, _constraint=f.constraint)
- for f in self.foreign_keys
+ ForeignKey(
+ col if col is not None else f._colspec,
+ _unresolvable=col is None,
+ _constraint=f.constraint,
+ )
+ for f, col in [
+ (fk, fk._resolve_column(raiseerr=False))
+ for fk in self.foreign_keys
+ ]
]
+
if name is None and self.name is None:
raise exc.InvalidRequestError(
"Cannot initialize a sub-selectable"
link_to_name=False,
match=None,
info=None,
+ _unresolvable=False,
**dialect_kw
):
r"""
"""
self._colspec = coercions.expect(roles.DDLReferredColumnRole, column)
+ self._unresolvable = _unresolvable
if isinstance(self._colspec, util.string_types):
self._table_column = None
parenttable = self.parent.table
+ if self._unresolvable:
+ schema, tname, colname = self._column_tokens
+ tablekey = _get_table_key(tname, schema)
+ return parenttable, tablekey, colname
+
# assertion
# basically Column._make_proxy() sends the actual
# target Column to the ForeignKey object, so the
"""
+ return self._resolve_column()
+
+ def _resolve_column(self, raiseerr=True):
+
if isinstance(self._colspec, util.string_types):
parenttable, tablekey, colname = self._resolve_col_tokens()
- if tablekey not in parenttable.metadata:
+ if self._unresolvable or tablekey not in parenttable.metadata:
+ if not raiseerr:
+ return None
raise exc.NoReferencedTableError(
"Foreign key associated with column '%s' could not find "
"table '%s' with which to generate a "
tablekey,
)
elif parenttable.key not in parenttable.metadata:
+ if not raiseerr:
+ return None
raise exc.InvalidRequestError(
"Table %s is no longer associated with its "
"parent MetaData" % parenttable
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import schema
+from sqlalchemy import select
from sqlalchemy import sql
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import eq_regex
+from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import in_
)
assert "nonexistent" not in meta.tables
- def test_include_columns(self, connection, metadata):
- meta = metadata
- foo = Table(
- "foo",
- meta,
- *[Column(n, sa.String(30)) for n in ["a", "b", "c", "d", "e", "f"]]
- )
- meta.create_all(connection)
- meta2 = MetaData()
- foo = Table(
- "foo",
- meta2,
- autoload_with=connection,
- include_columns=["b", "f", "e"],
- )
- # test that cols come back in original order
- eq_([c.name for c in foo.c], ["b", "e", "f"])
- for c in ("b", "f", "e"):
- assert c in foo.c
- for c in ("a", "c", "d"):
- assert c not in foo.c
-
- # test against a table which is already reflected
- meta3 = MetaData()
- foo = Table("foo", meta3, autoload_with=connection)
-
- foo = Table(
- "foo", meta3, include_columns=["b", "f", "e"], extend_existing=True
- )
- eq_([c.name for c in foo.c], ["b", "e", "f"])
- for c in ("b", "f", "e"):
- assert c in foo.c
- for c in ("a", "c", "d"):
- assert c not in foo.c
-
def test_extend_existing(self, connection, metadata):
meta = metadata
is_true(table.c.id1.identity is not None)
eq_(table.c.id1.identity.start, 2)
eq_(table.c.id1.identity.increment, 3)
+
+
+class IncludeColsFksTest(AssertsCompiledSQL, fixtures.TestBase):
+ __dialect__ = "default"
+
+ @testing.fixture
+ def tab_wo_fks(self, connection, metadata):
+ meta = metadata
+ foo = Table(
+ "foo",
+ meta,
+ *[Column(n, sa.String(30)) for n in ["a", "b", "c", "d", "e", "f"]]
+ )
+ meta.create_all(connection)
+
+ return foo
+
+ @testing.fixture
+ def tab_w_fks(self, connection, metadata):
+ Table(
+ "a",
+ metadata,
+ Column("x", Integer, primary_key=True),
+ test_needs_fk=True,
+ )
+
+ b = Table(
+ "b",
+ metadata,
+ Column("x", Integer, primary_key=True),
+ Column("q", Integer),
+ Column("p", Integer),
+ Column("r", Integer, ForeignKey("a.x")),
+ Column("s", Integer),
+ Column("t", Integer),
+ test_needs_fk=True,
+ )
+
+ metadata.create_all(connection)
+
+ return b
+
+ def test_include_columns(self, connection, tab_wo_fks):
+ foo = tab_wo_fks
+ meta2 = MetaData()
+ foo = Table(
+ "foo",
+ meta2,
+ autoload_with=connection,
+ include_columns=["b", "f", "e"],
+ )
+ # test that cols come back in original order
+ eq_([c.name for c in foo.c], ["b", "e", "f"])
+ for c in ("b", "f", "e"):
+ assert c in foo.c
+ for c in ("a", "c", "d"):
+ assert c not in foo.c
+
+ # test against a table which is already reflected
+ meta3 = MetaData()
+ foo = Table("foo", meta3, autoload_with=connection)
+
+ foo = Table(
+ "foo", meta3, include_columns=["b", "f", "e"], extend_existing=True
+ )
+ eq_([c.name for c in foo.c], ["b", "e", "f"])
+ for c in ("b", "f", "e"):
+ assert c in foo.c
+ for c in ("a", "c", "d"):
+ assert c not in foo.c
+
+ @testing.emits_warning
+ @testing.combinations(True, False, argnames="resolve_fks")
+ def test_include_cols_skip_fk_col(
+ self, connection, tab_w_fks, resolve_fks
+ ):
+ """test #8100"""
+
+ m2 = MetaData()
+
+ b2 = Table(
+ "b",
+ m2,
+ autoload_with=connection,
+ resolve_fks=resolve_fks,
+ include_columns=["x", "q", "p"],
+ )
+
+ eq_([c.name for c in b2.c], ["x", "q", "p"])
+
+ # no FK, whether or not resolve_fks was called
+ eq_(b2.constraints, set((b2.primary_key,)))
+
+ b2a = b2.alias()
+ eq_([c.name for c in b2a.c], ["x", "q", "p"])
+
+ self.assert_compile(select(b2), "SELECT b.x, b.q, b.p FROM b")
+ self.assert_compile(
+ select(b2.alias()),
+ "SELECT b_1.x, b_1.q, b_1.p FROM b AS b_1",
+ )
+
+ def test_table_works_minus_fks(self, connection, tab_w_fks):
+ """test #8101"""
+
+ m2 = MetaData()
+
+ b2 = Table(
+ "b",
+ m2,
+ autoload_with=connection,
+ resolve_fks=False,
+ )
+
+ eq_([c.name for c in b2.c], ["x", "q", "p", "r", "s", "t"])
+
+ b2a = b2.alias()
+ eq_([c.name for c in b2a.c], ["x", "q", "p", "r", "s", "t"])
+
+ self.assert_compile(
+ select(b2), "SELECT b.x, b.q, b.p, b.r, b.s, b.t FROM b"
+ )
+ b2a_1 = b2.alias()
+ self.assert_compile(
+ select(b2a_1),
+ "SELECT b_1.x, b_1.q, b_1.p, b_1.r, b_1.s, b_1.t FROM b AS b_1",
+ )
+
+ # reflecting the related table
+ a2 = Table("a", m2, autoload_with=connection)
+
+ # the existing alias doesn't know about it
+ with expect_raises_message(
+ sa.exc.InvalidRequestError,
+ "Foreign key associated with column 'anon_1.r' could not find "
+ "table 'a' with which to generate a foreign key to target "
+ "column 'x'",
+ ):
+ select(b2a_1).join(a2).compile()
+
+ # can still join manually (needed to fix inside of util for this...)
+ self.assert_compile(
+ select(b2a_1).join(a2, b2a_1.c.r == a2.c.x),
+ "SELECT b_1.x, b_1.q, b_1.p, b_1.r, b_1.s, b_1.t "
+ "FROM b AS b_1 JOIN a ON b_1.r = a.x",
+ )
+
+ # a new alias does know about it however
+ self.assert_compile(
+ select(b2.alias()).join(a2),
+ "SELECT b_1.x, b_1.q, b_1.p, b_1.r, b_1.s, b_1.t "
+ "FROM b AS b_1 JOIN a ON a.x = b_1.r",
+ )
assert j4.corresponding_column(j2.c.aid) is j4.c.aid
assert j4.corresponding_column(a.c.id) is j4.c.id
- def test_two_metadata_join_raises(self):
+ @testing.combinations(True, False)
+ def test_two_metadata_join_raises(self, include_a_joining_table):
+ """test case from 2008 enhanced as of #8101, more specific failure
+ modes for non-resolvable FKs
+
+ """
m = MetaData()
m2 = MetaData()
t1 = Table("t1", m, Column("id", Integer), Column("id2", Integer))
- t2 = Table("t2", m, Column("id", Integer, ForeignKey("t1.id")))
+
+ if include_a_joining_table:
+ t2 = Table("t2", m, Column("id", Integer, ForeignKey("t1.id")))
+
t3 = Table("t3", m2, Column("id", Integer, ForeignKey("t1.id2")))
- s = (
- select(t2, t3)
- .set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
- .subquery()
- )
+ with expect_raises_message(
+ exc.NoReferencedTableError,
+ "Foreign key associated with column 't3.id'",
+ ):
+ t3.join(t1)
- assert_raises(exc.NoReferencedTableError, s.join, t1)
+ if include_a_joining_table:
+ s = (
+ select(t2, t3)
+ .set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
+ .subquery()
+ )
+ else:
+ s = (
+ select(t3)
+ .set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
+ .subquery()
+ )
+
+ with expect_raises_message(
+ exc.NoReferencedTableError,
+ "Foreign key associated with column 'anon_1.t3_id' could not "
+ "find table 't1' with which to generate a foreign key to target "
+ "column 'id2'",
+ ):
+ select(s.join(t1)),
+
+ # manual join is OK. using select().join() here is also exercising
+ # that join() does not need to resolve FKs if we provided the
+ # ON clause
+ if include_a_joining_table:
+ self.assert_compile(
+ select(s).join(
+ t1, and_(s.c.t2_id == t1.c.id, s.c.t3_id == t1.c.id)
+ ),
+ "SELECT anon_1.t2_id, anon_1.t3_id FROM (SELECT "
+ "t2.id AS t2_id, t3.id AS t3_id FROM t2, t3) AS anon_1 "
+ "JOIN t1 ON anon_1.t2_id = t1.id AND anon_1.t3_id = t1.id",
+ )
+ else:
+ self.assert_compile(
+ select(s).join(t1, s.c.t3_id == t1.c.id),
+ "SELECT anon_1.t3_id FROM (SELECT t3.id AS t3_id FROM t3) "
+ "AS anon_1 JOIN t1 ON anon_1.t3_id = t1.id",
+ )
def test_multi_label_chain_naming_col(self):
# See [ticket:2167] for this one.
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
