diff options
| -rw-r--r-- | doc/build/changelog/changelog_09.rst | 12 | ||||
| -rw-r--r-- | lib/sqlalchemy/sql/schema.py | 55 | ||||
| -rw-r--r-- | test/sql/test_metadata.py | 41 |
3 files changed, 86 insertions, 22 deletions
diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst index fa1006694..b1da813cb 100644 --- a/doc/build/changelog/changelog_09.rst +++ b/doc/build/changelog/changelog_09.rst @@ -15,6 +15,18 @@ :version: 0.9.0b2 .. change:: + :tags: bug, sql + :tickets: 2883 + + The :class:`.ForeignKey` class more aggressively checks the given + column argument. If not a string, it checks that the object is + at least a :class:`.ColumnClause`, or an object that resolves to one, + and that the ``.table`` attribute, if present, refers to a + :class:`.TableClause` or subclass, and not something like an + :class:`.Alias`. Otherwise, a :class:`.ArgumentError` is raised. + + + .. change:: :tags: feature, orm The :class:`.exc.StatementError` or DBAPI-related subclass diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py index 7bf543a61..4d9dc2bda 100644 --- a/lib/sqlalchemy/sql/schema.py +++ b/lib/sqlalchemy/sql/schema.py @@ -1340,6 +1340,23 @@ class ForeignKey(SchemaItem): """ self._colspec = column + if isinstance(self._colspec, util.string_types): + self._table_column = None + else: + if hasattr(self._colspec, '__clause_element__'): + self._table_column = self._colspec.__clause_element__() + else: + self._table_column = self._colspec + + if not isinstance(self._table_column, ColumnClause): + raise exc.ArgumentError( + "String, Column, or Column-bound argument " + "expected, got %r" % self._table_column) + elif not isinstance(self._table_column.table, (util.NoneType, TableClause)): + raise exc.ArgumentError( + "ForeignKey received Column not bound " + "to a Table, got: %r" % self._table_column.table + ) # the linked ForeignKeyConstraint. # ForeignKey will create this when parent Column @@ -1389,6 +1406,7 @@ class ForeignKey(SchemaItem): ) return self._schema_item_copy(fk) + def _get_colspec(self, schema=None): """Return a string based 'column specification' for this :class:`.ForeignKey`. @@ -1398,16 +1416,25 @@ class ForeignKey(SchemaItem): """ if schema: - return schema + "." + self.column.table.name + \ - "." + self.column.key - elif isinstance(self._colspec, util.string_types): + _schema, tname, colname = self._column_tokens + return "%s.%s.%s" % (schema, tname, colname) + elif self._table_column is not None: + return "%s.%s" % ( + self._table_column.table.fullname, self._table_column.key) + else: return self._colspec - elif hasattr(self._colspec, '__clause_element__'): - _column = self._colspec.__clause_element__() + + + def _table_key(self): + if self._table_column is not None: + if self._table_column.table is None: + return None + else: + return self._table_column.table.key else: - _column = self._colspec + schema, tname, colname = self._column_tokens + return _get_table_key(tname, schema) - return "%s.%s" % (_column.table.fullname, _column.key) target_fullname = property(_get_colspec) @@ -1460,20 +1487,6 @@ class ForeignKey(SchemaItem): schema = None return schema, tname, colname - def _table_key(self): - if isinstance(self._colspec, util.string_types): - schema, tname, colname = self._column_tokens - return _get_table_key(tname, schema) - elif hasattr(self._colspec, '__clause_element__'): - _column = self._colspec.__clause_element__() - else: - _column = self._colspec - - if _column.table is None: - return None - else: - return _column.table.key - def _resolve_col_tokens(self): if self.parent is None: raise exc.InvalidRequestError( diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py index d0a79a7bb..c5caa9780 100644 --- a/test/sql/test_metadata.py +++ b/test/sql/test_metadata.py @@ -6,7 +6,7 @@ import pickle from sqlalchemy import Integer, String, UniqueConstraint, \ CheckConstraint, ForeignKey, MetaData, Sequence, \ ForeignKeyConstraint, ColumnDefault, Index, event,\ - events, Unicode, types as sqltypes + events, Unicode, types as sqltypes, bindparam from sqlalchemy.testing.schema import Table, Column from sqlalchemy import schema, exc import sqlalchemy as tsa @@ -236,6 +236,45 @@ class MetaDataTest(fixtures.TestBase, ComparesTables): go ) + def test_fk_given_non_col(self): + not_a_col = bindparam('x') + assert_raises_message( + exc.ArgumentError, + "String, Column, or Column-bound argument expected, got Bind", + ForeignKey, not_a_col + ) + + def test_fk_given_non_col_clauseelem(self): + class Foo(object): + def __clause_element__(self): + return bindparam('x') + assert_raises_message( + exc.ArgumentError, + "String, Column, or Column-bound argument expected, got Bind", + ForeignKey, Foo() + ) + + def test_fk_given_col_non_table(self): + t = Table('t', MetaData(), Column('x', Integer)) + xa = t.alias().c.x + assert_raises_message( + exc.ArgumentError, + "ForeignKey received Column not bound to a Table, got: .*Alias", + ForeignKey, xa + ) + + def test_fk_given_col_non_table_clauseelem(self): + t = Table('t', MetaData(), Column('x', Integer)) + class Foo(object): + def __clause_element__(self): + return t.alias().c.x + + assert_raises_message( + exc.ArgumentError, + "ForeignKey received Column not bound to a Table, got: .*Alias", + ForeignKey, Foo() + ) + def test_fk_no_such_target_col_error_upfront(self): meta = MetaData() a = Table('a', meta, Column('a', Integer)) |