also would be a little misleading compared to
values().
+ - Reflected foreign keys will properly locate
+ their referenced column, even if the column
+ was given a "key" attribute different from
+ the reflected name. This is achieved via a
+ new flag on ForeignKey/ForeignKeyConstraint
+ called "link_to_name", if True means the given
+ name is the referred-to column's name, not its
+ assigned key.
+ [ticket:650]
+
- select() can accept a ClauseList as a column
in the same way as a Table or other selectable
and the interior expressions will be used as
.. automodule:: sqlalchemy.databases.sqlite
-SQLite Column Types
-------------------
-
-.. autoclass:: SLBinary
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLBoolean
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLChar
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLDateTime
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLDate
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLFloat
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLNumeric
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLInteger
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLSmallInteger
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLString
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLText
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: SLTime
- :members: __init__
- :show-inheritance:
-
-.. autoclass:: DateTimeMixin
- :members:
- :show-inheritance:
-
-.. autoclass:: SLUnicodeMixin
- :members:
- :show-inheritance:
-
continue
scols = [c.ForeignName for c in fk.Fields]
rcols = ['%s.%s' % (fk.Table, c.Name) for c in fk.Fields]
- table.append_constraint(schema.ForeignKeyConstraint(scols, rcols))
+ table.append_constraint(schema.ForeignKeyConstraint(scols, rcols, link_to_name=True))
finally:
dtbs.Close()
fk[1].append(refspec)
for name, value in fks.iteritems():
- table.append_constraint(schema.ForeignKeyConstraint(value[0], value[1], name=name))
+ table.append_constraint(schema.ForeignKeyConstraint(value[0], value[1], name=name, link_to_name=True))
def do_execute(self, cursor, statement, parameters, **kwargs):
# kinterbase does not accept a None, but wants an empty list
fk[1].append(refspec)
for name, value in fks.iteritems():
- table.append_constraint(schema.ForeignKeyConstraint(value[0], value[1] , None ))
+ table.append_constraint(schema.ForeignKeyConstraint(value[0], value[1] , None, link_to_name=True ))
# PK
c = connection.execute("""select t1.constrname as cons_name , t1.constrtype as cons_type ,
autoload=True, autoload_with=connection,
**table_kw)
- constraint = schema.ForeignKeyConstraint(columns, referants,
+ constraint = schema.ForeignKeyConstraint(columns, referants, link_to_name=True,
**constraint_kw)
table.append_constraint(constraint)
schema.Table(rtbl, table.metadata, schema=rschema, autoload=True, autoload_with=connection)
if rfknm != fknm:
if fknm:
- table.append_constraint(schema.ForeignKeyConstraint(scols, [_gen_fkref(table, s, t, c) for s, t, c in rcols], fknm))
+ table.append_constraint(schema.ForeignKeyConstraint(scols, [_gen_fkref(table, s, t, c) for s, t, c in rcols], fknm, link_to_name=True))
fknm, scols, rcols = (rfknm, [], [])
if not scol in scols:
scols.append(scol)
rcols.append((rschema, rtbl, rcol))
if fknm and scols:
- table.append_constraint(schema.ForeignKeyConstraint(scols, [_gen_fkref(table, s, t, c) for s, t, c in rcols], fknm))
+ table.append_constraint(schema.ForeignKeyConstraint(scols, [_gen_fkref(table, s, t, c) for s, t, c in rcols], fknm, link_to_name=True))
class MSSQLDialect_pymssql(MSSQLDialect):
autoload=True, autoload_with=connection)
ref_names = spec['foreign']
- if not set(ref_names).issubset(
- set(c.name for c in ref_table.c)):
- raise exc.InvalidRequestError(
- "Foreign key columns (%s) are not present on "
- "foreign table %s" %
- (', '.join(ref_names), ref_table.fullname))
- ref_columns = [ref_table.c[name] for name in ref_names]
+
+ if ref_schema:
+ refspec = [".".join([ref_schema, ref_name, column]) for column in ref_names]
+ else:
+ refspec = [".".join([ref_name, column]) for column in ref_names]
con_kw = {}
for opt in ('name', 'onupdate', 'ondelete'):
if spec.get(opt, False):
con_kw[opt] = spec[opt]
- key = schema.ForeignKeyConstraint([], [], **con_kw)
+ key = schema.ForeignKeyConstraint(loc_names, refspec, link_to_name=True, **con_kw)
table.append_constraint(key)
- for pair in zip(loc_names, ref_columns):
- key.append_element(*pair)
def _set_options(self, table, line):
"""Apply safe reflected table options to a ``Table``.
fk[1].append(refspec)
for name, value in fks.iteritems():
- table.append_constraint(schema.ForeignKeyConstraint(value[0], value[1], name=name))
+ table.append_constraint(schema.ForeignKeyConstraint(value[0], value[1], name=name, link_to_name=True))
class _OuterJoinColumn(sql.ClauseElement):
c = connection.execute(t, table=table_oid)
for row in c.fetchall():
pk = row[0]
- col = table.c[pk]
- table.primary_key.add(col)
- if col.default is None:
- col.autoincrement = False
+ if pk in table.c:
+ col = table.c[pk]
+ table.primary_key.add(col)
+ if col.default is None:
+ col.autoincrement = False
# Foreign keys
FK_SQL = """
for column in referred_columns:
refspec.append(".".join([referred_table, column]))
- table.append_constraint(schema.ForeignKeyConstraint(constrained_columns, refspec, conname))
+ table.append_constraint(schema.ForeignKeyConstraint(constrained_columns, refspec, conname, link_to_name=True))
# Indexes
IDX_SQL = """
if refspec not in fk[1]:
fk[1].append(refspec)
for name, value in fks.iteritems():
- table.append_constraint(schema.ForeignKeyConstraint(value[0], value[1]))
+ table.append_constraint(schema.ForeignKeyConstraint(value[0], value[1], link_to_name=True))
# check for UNIQUE indexes
c = connection.execute("%sindex_list(%s)" % (pragma, qtable))
unique_indexes = []
foreignKeys[primary_table][1].append('%s.%s'%(primary_table, primary_column))
for primary_table in foreignKeys.keys():
#table.append_constraint(schema.ForeignKeyConstraint(['%s.%s'%(foreign_table, foreign_column)], ['%s.%s'%(primary_table,primary_column)]))
- table.append_constraint(schema.ForeignKeyConstraint(foreignKeys[primary_table][0], foreignKeys[primary_table][1]))
+ table.append_constraint(schema.ForeignKeyConstraint(foreignKeys[primary_table][0], foreignKeys[primary_table][1], link_to_name=True))
if not found_table:
raise exc.NoSuchTableError(table.name)
__visit_name__ = 'foreign_key'
- def __init__(self, column, constraint=None, use_alter=False, name=None, onupdate=None, ondelete=None, deferrable=None, initially=None):
+ def __init__(self, column, constraint=None, use_alter=False, name=None, onupdate=None, ondelete=None, deferrable=None, initially=None, link_to_name=False):
"""
Construct a column-level FOREIGN KEY.
:param column: A single target column for the key relationship. A :class:`Column`
- object or a column name as a string: ``tablename.columnname`` or
- ``schema.tablename.columnname``.
+ object or a column name as a string: ``tablename.columnkey`` or
+ ``schema.tablename.columnkey``. ``columnkey`` is the ``key`` which has been assigned
+ to the column (defaults to the column name itself), unless ``link_to_name`` is ``True``
+ in which case the rendered name of the column is used.
:param constraint: Optional. A parent :class:`ForeignKeyConstraint` object. If not
supplied, a :class:`ForeignKeyConstraint` will be automatically created
:param initially: Optional string. If set, emit INITIALLY <value> when issuing DDL
for this constraint.
-
+
+ :param link_to_name: if True, the string name given in ``column`` is the rendered
+ name of the referenced column, not its locally assigned ``key``.
+
:param use_alter: If True, do not emit this key as part of the CREATE TABLE
definition. Instead, use ALTER TABLE after table creation to add
the key. Useful for circular dependencies.
self.ondelete = ondelete
self.deferrable = deferrable
self.initially = initially
+ self.link_to_name = link_to_name
def __repr__(self):
return "ForeignKey(%r)" % self._get_colspec()
"foreign key" % tname)
table = Table(tname, parenttable.metadata,
mustexist=True, schema=schema)
- try:
- if colname is None:
- # colname is None in the case that ForeignKey argument
- # was specified as table name only, in which case we
- # match the column name to the same column on the
- # parent.
- key = self.parent
- _column = table.c[self.parent.key]
- else:
- _column = table.c[colname]
- except KeyError, e:
+
+ _column = None
+ if colname is None:
+ # colname is None in the case that ForeignKey argument
+ # was specified as table name only, in which case we
+ # match the column name to the same column on the
+ # parent.
+ key = self.parent
+ _column = table.c.get(self.parent.key, None)
+ elif self.link_to_name:
+ key = colname
+ for c in table.c:
+ if c.name == colname:
+ _column = c
+ else:
+ key = colname
+ _column = table.c.get(colname, None)
+
+ if not _column:
raise exc.NoReferencedColumnError(
"Could not create ForeignKey '%s' on table '%s': "
"table '%s' has no column named '%s'" % (
- self._colspec, parenttable.name, table.name, str(e)))
+ self._colspec, parenttable.name, table.name, key))
elif hasattr(self._colspec, '__clause_element__'):
_column = self._colspec.__clause_element__()
"""
__visit_name__ = 'foreign_key_constraint'
- def __init__(self, columns, refcolumns, name=None, onupdate=None, ondelete=None, use_alter=False, deferrable=None, initially=None):
+ def __init__(self, columns, refcolumns, name=None, onupdate=None, ondelete=None, use_alter=False, deferrable=None, initially=None, link_to_name=False):
"""Construct a composite-capable FOREIGN KEY.
- columns
- A sequence of local column names. The named columns must be defined
- and present in the parent Table.
+ :param columns: A sequence of local column names. The named columns must be defined
+ and present in the parent Table. The names should match the ``key`` given
+ to each column (defaults to the name) unless ``link_to_name`` is True.
- refcolumns
- A sequence of foreign column names or Column objects. The columns
+ :param refcolumns: A sequence of foreign column names or Column objects. The columns
must all be located within the same Table.
- name
- Optional, the in-database name of the key.
+ :param name: Optional, the in-database name of the key.
- onupdate
- Optional string. If set, emit ON UPDATE <value> when issuing DDL
+ :param onupdate: Optional string. If set, emit ON UPDATE <value> when issuing DDL
for this constraint. Typical values include CASCADE, DELETE and
RESTRICT.
- ondelete
- Optional string. If set, emit ON DELETE <value> when issuing DDL
+ :param ondelete: Optional string. If set, emit ON DELETE <value> when issuing DDL
for this constraint. Typical values include CASCADE, DELETE and
RESTRICT.
- deferrable
- Optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when
+ :param deferrable: Optional bool. If set, emit DEFERRABLE or NOT DEFERRABLE when
issuing DDL for this constraint.
- initially
- Optional string. If set, emit INITIALLY <value> when issuing DDL
+ :param initially: Optional string. If set, emit INITIALLY <value> when issuing DDL
for this constraint.
- use_alter
- If True, do not emit this key as part of the CREATE TABLE
+ :param link_to_name: if True, the string name given in ``column`` is the rendered
+ name of the referenced column, not its locally assigned ``key``.
+
+ :param use_alter: If True, do not emit this key as part of the CREATE TABLE
definition. Instead, use ALTER TABLE after table creation to add
the key. Useful for circular dependencies.
+
"""
-
super(ForeignKeyConstraint, self).__init__(name, deferrable, initially)
self.__colnames = columns
self.__refcolnames = refcolumns
self.elements = util.OrderedSet()
self.onupdate = onupdate
self.ondelete = ondelete
+ self.link_to_name = link_to_name
if self.name is None and use_alter:
raise exc.ArgumentError("Alterable ForeignKey/ForeignKeyConstraint requires a name")
self.use_alter = use_alter
self.append_element(c, r)
def append_element(self, col, refcol):
- fk = ForeignKey(refcol, constraint=self, name=self.name, onupdate=self.onupdate, ondelete=self.ondelete, use_alter=self.use_alter)
+ fk = ForeignKey(refcol, constraint=self, name=self.name, onupdate=self.onupdate, ondelete=self.ondelete, use_alter=self.use_alter, link_to_name=self.link_to_name)
fk._set_parent(self.table.c[col])
self._append_fk(fk)
t.create()
dialect_module.ischema_names = {}
try:
- try:
- m2 = MetaData(testing.db)
- t2 = Table("test", m2, autoload=True)
- assert False
- except tsa.exc.SAWarning:
- assert True
+ m2 = MetaData(testing.db)
+ self.assertRaises(tsa.exc.SAWarning, Table, "test", m2, autoload=True)
@testing.emits_warning('Did not recognize type')
def warns():
finally:
meta.drop_all()
+ def test_override_keys(self):
+ """test that columns can be overridden with a 'key',
+ and that ForeignKey targeting during reflection still works."""
+
+
+ meta = MetaData(testing.db)
+ a1 = Table('a', meta,
+ Column('x', sa.Integer, primary_key=True),
+ Column('z', sa.Integer),
+ test_needs_fk=True
+ )
+ b1 = Table('b', meta,
+ Column('y', sa.Integer, sa.ForeignKey('a.x')),
+ test_needs_fk=True
+ )
+ meta.create_all()
+ try:
+ m2 = MetaData(testing.db)
+ a2 = Table('a', m2, Column('x', sa.Integer, primary_key=True, key='x1'), autoload=True)
+ b2 = Table('b', m2, autoload=True)
+
+ assert a2.join(b2).onclause.compare(a2.c.x1==b2.c.y)
+ assert b2.c.y.references(a2.c.x1)
+ finally:
+ meta.drop_all()
+
+ def test_nonreflected_fk_raises(self):
+ """test that a NoReferencedColumnError is raised when reflecting
+ a table with an FK to another table which has not included the target
+ column in its reflection.
+
+ """
+ meta = MetaData(testing.db)
+ a1 = Table('a', meta,
+ Column('x', sa.Integer, primary_key=True),
+ Column('z', sa.Integer),
+ test_needs_fk=True
+ )
+ b1 = Table('b', meta,
+ Column('y', sa.Integer, sa.ForeignKey('a.x')),
+ test_needs_fk=True
+ )
+ meta.create_all()
+ try:
+ m2 = MetaData(testing.db)
+ a2 = Table('a', m2, include_columns=['z'], autoload=True)
+ b2 = Table('b', m2, autoload=True)
+
+ self.assertRaises(tsa.exc.NoReferencedColumnError, a2.join, b2)
+ finally:
+ meta.drop_all()
+
+
@testing.exclude('mysql', '<', (4, 1, 1), 'innodb funkiness')
def test_override_existing_fk(self):
"""test that you can override columns and specify new foreign keys to other reflected tables,
Column('pkg_id', sa.Integer, sa.ForeignKey('pkgs.pkg_id')),
Column('slot', sa.String(128)),
)
- try:
- metadata.create_all()
- assert False
- except tsa.exc.InvalidRequestError, err:
- assert str(err) == "Could not find table 'pkgs' with which to generate a foreign key"
+
+ self.assertRaisesMessage(tsa.exc.InvalidRequestError, "Could not find table 'pkgs' with which to generate a foreign key", metadata.create_all)
def test_composite_pks(self):
"""test reflection of a composite primary key"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
