- simplify the mechanics of PrimaryKeyConstraint with regards to reflection;

reflection now updates the PKC in place.
- support the use case of the empty PrimaryKeyConstraint in order to specify
constraint options; the columns marked as primary_key=True will now be gathered
into the columns collection, rather than being ignored. [ticket:2910]
- add validation such that column specification should only take place
in the PrimaryKeyConstraint directly, or by using primary_key=True flags;
if both are present, they have to match exactly, otherwise the condition is
assumed to be ambiguous, and a warning is emitted; the old behavior of
using the PKC columns only is maintained.

7 files changed, 267 insertions, 25 deletions

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index 63f95d242..0efffce62 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -16,6 +16,33 @@
 
     .. change::
         :tags: feature, sql
+        :tickets: 2910
+
+        Options can now be specified on a :class:`.PrimaryKeyConstraint` object
+        independently of the specification of columns in the table with
+        the ``primary_key=True`` flag; use a :class:`.PrimaryKeyConstraint`
+        object with no columns in it to achieve this result.
+
+        Previously, an explicit :class:`.PrimaryKeyConstraint` would have the
+        effect of those columns marked as ``primary_key=True`` being ignored;
+        since this is no longer the case, the :class:`.PrimaryKeyConstraint`
+        will now assert that either one style or the other is used to specify
+        the columns, or if both are present, that the column lists match
+        exactly.  If an inconsistent set of columns in the
+        :class:`.PrimaryKeyConstraint`
+        and within the :class:`.Table` marked as ``primary_key=True`` are
+        present, a warning is emitted, and the list of columns is taken
+        only from the :class:`.PrimaryKeyConstraint` alone as was the case
+        in previous releases.
+
+
+
+        .. seealso::
+
+            :class:`.PrimaryKeyConstraint`
+
+    .. change::
+        :tags: feature, sql
         :tickets: 2866
 
         The system by which schema constructs and certain SQL constructs
diff --git a/doc/build/core/constraints.rst b/doc/build/core/constraints.rst
index 75150cbd9..13ead6fbf 100644
--- a/doc/build/core/constraints.rst
+++ b/doc/build/core/constraints.rst
@@ -178,6 +178,8 @@ unique constraints and/or those with multiple columns are created via the
 
 .. sourcecode:: python+sql
 
+    from sqlalchemy import UniqueConstraint
+
     meta = MetaData()
     mytable = Table('mytable', meta,
 
@@ -206,6 +208,8 @@ MySQL.
 
 .. sourcecode:: python+sql
 
+    from sqlalchemy import CheckConstraint
+
     meta = MetaData()
     mytable = Table('mytable', meta,
 
@@ -227,6 +231,28 @@ MySQL.
         CONSTRAINT check1  CHECK (col2 > col3 + 5)
     ){stop}
 
+PRIMARY KEY Constraint
+----------------------
+
+The primary key constraint of any :class:`.Table` object is implicitly
+present, based on the :class:`.Column` objects that are marked with the
+:paramref:`.Column.primary_key` flag.   The :class:`.PrimaryKeyConstraint`
+object provides explicit access to this constraint, which includes the
+option of being configured directly::
+
+    from sqlalchemy import PrimaryKeyConstraint
+
+    my_table = Table('mytable', metadata,
+                Column('id', Integer),
+                Column('version_id', Integer),
+                Column('data', String(50)),
+                PrimaryKeyConstraint('id', 'version_id', name='mytable_pk')
+            )
+
+.. seealso::
+
+    :class:`.PrimaryKeyConstraint` - detailed API documentation.
+
 Setting up Constraints when using the Declarative ORM Extension
 ----------------------------------------------------------------
 
diff --git a/doc/build/requirements.txt b/doc/build/requirements.txt
index 6e2354f07..b7881327d 100644
--- a/doc/build/requirements.txt
+++ b/doc/build/requirements.txt
@@ -1,3 +1,3 @@
 mako
 changelog>=0.3.4
-sphinx-paramlinks>=0.2.0
+sphinx-paramlinks>=0.2.1
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index d82aac7fd..9e6cf61dc 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -504,6 +504,8 @@ class Inspector(object):
             cols_by_orig_name[orig_name] = col = \
                         sa_schema.Column(name, coltype, *colargs, **col_kw)
 
+            if col.key in table.primary_key:
+                col.primary_key = True
             table.append_column(col)
 
         if not found_table:
@@ -516,17 +518,20 @@ class Inspector(object):
                 for pk in pk_cons['constrained_columns']
                 if pk in cols_by_orig_name and pk not in exclude_columns
             ]
-            pk_cols += [
-                pk
-                for pk in table.primary_key
-                if pk.key in exclude_columns
-            ]
-            primary_key_constraint = sa_schema.PrimaryKeyConstraint(
-                name=pk_cons.get('name'),
-                *pk_cols
-            )
 
-            table.append_constraint(primary_key_constraint)
+            # update pk constraint name
+            table.primary_key.name = pk_cons.get('name')
+
+            # set the primary key flag on new columns.
+            # note any existing PK cols on the table also have their
+            # flag still set.
+            for col in pk_cols:
+                col.primary_key = True
+
+            # tell the PKConstraint to re-initialize
+            # it's column collection
+            table.primary_key._reload()
+
 
         fkeys = self.get_foreign_keys(table_name, schema, **table.dialect_kwargs)
         for fkey_d in fkeys:
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 73c2a49c8..09f52f8a7 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -428,11 +428,6 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
 
     def _autoload(self, metadata, autoload_with, include_columns,
                   exclude_columns=()):
-        if self.primary_key.columns:
-            PrimaryKeyConstraint(*[
-                c for c in self.primary_key.columns
-                if c.key in exclude_columns
-            ])._set_parent_with_dispatch(self)
 
         if autoload_with:
             autoload_with.run_callable(
@@ -2532,10 +2527,69 @@ class ForeignKeyConstraint(Constraint):
 class PrimaryKeyConstraint(ColumnCollectionConstraint):
     """A table-level PRIMARY KEY constraint.
 
-    Defines a single column or composite PRIMARY KEY constraint. For a
-    no-frills primary key, adding ``primary_key=True`` to one or more
-    ``Column`` definitions is a shorthand equivalent for an unnamed single- or
-    multiple-column PrimaryKeyConstraint.
+    The :class:`.PrimaryKeyConstraint` object is present automatically
+    on any :class:`.Table` object; it is assigned a set of
+    :class:`.Column` objects corresponding to those marked with
+    the :paramref:`.Column.primary_key` flag::
+
+        >>> my_table = Table('mytable', metadata,
+        ...                 Column('id', Integer, primary_key=True),
+        ...                 Column('version_id', Integer, primary_key=True),
+        ...                 Column('data', String(50))
+        ...     )
+        >>> my_table.primary_key
+        PrimaryKeyConstraint(
+            Column('id', Integer(), table=<mytable>, primary_key=True, nullable=False),
+            Column('version_id', Integer(), table=<mytable>, primary_key=True, nullable=False)
+        )
+
+    The primary key of a :class:`.Table` can also be specified by using
+    a :class:`.PrimaryKeyConstraint` object explicitly; in this mode of usage,
+    the "name" of the constraint can also be specified, as well as other
+    options which may be recognized by dialects::
+
+        my_table = Table('mytable', metadata,
+                    Column('id', Integer),
+                    Column('version_id', Integer),
+                    Column('data', String(50)),
+                    PrimaryKeyConstraint('id', 'version_id', name='mytable_pk')
+                )
+
+    The two styles of column-specification should generally not be mixed.
+    An warning is emitted if the columns present in the
+    :class:`.PrimaryKeyConstraint`
+    don't match the columns that were marked as ``primary_key=True``, if both
+    are present; in this case, the columns are taken strictly from the
+    :class:`.PrimaryKeyConstraint` declaration, and those columns otherwise marked
+    as ``primary_key=True`` are ignored.  This behavior is intended to be
+    backwards compatible with previous behavior.
+
+    .. versionchanged:: 0.9.2  Using a mixture of columns within a
+       :class:`.PrimaryKeyConstraint` in addition to columns marked as
+       ``primary_key=True`` now emits a warning if the lists don't match.
+       The ultimate behavior of ignoring those columns marked with the flag
+       only is currently maintained for backwards compatibility; this warning
+       may raise an exception in a future release.
+
+    For the use case where specific options are to be specified on the
+    :class:`.PrimaryKeyConstraint`, but the usual style of using ``primary_key=True``
+    flags is still desirable, an empty :class:`.PrimaryKeyConstraint` may be
+    specified, which will take on the primary key column collection from
+    the :class:`.Table` based on the flags::
+
+        my_table = Table('mytable', metadata,
+                    Column('id', Integer, primary_key=True),
+                    Column('version_id', Integer, primary_key=True),
+                    Column('data', String(50)),
+                    PrimaryKeyConstraint(name='mytable_pk', mssql_clustered=True)
+                )
+
+    .. versionadded:: 0.9.2 an empty :class:`.PrimaryKeyConstraint` may now
+       be specified for the purposes of establishing keyword arguments with the
+       constraint, independently of the specification of "primary key" columns
+       within the :class:`.Table` itself; columns marked as ``primary_key=True``
+       will be gathered into the empty constraint's column collection.
+
     """
 
     __visit_name__ = 'primary_key_constraint'
@@ -2543,13 +2597,51 @@ class PrimaryKeyConstraint(ColumnCollectionConstraint):
     def _set_parent(self, table):
         super(PrimaryKeyConstraint, self)._set_parent(table)
 
-        if table.primary_key in table.constraints:
-            table.constraints.remove(table.primary_key)
-        table.primary_key = self
-        table.constraints.add(self)
+        if table.primary_key is not self:
+            table.constraints.discard(table.primary_key)
+            table.primary_key = self
+            table.constraints.add(self)
+
+        table_pks = [c for c in table.c if c.primary_key]
+        if self.columns and table_pks and \
+            set(table_pks) != set(self.columns.values()):
+            util.warn(
+                    "Table '%s' specifies columns %s as primary_key=True, "
+                    "not matching locally specified columns %s; setting the "
+                    "current primary key columns to %s. This warning "
+                    "may become an exception in a future release" %
+                    (
+                        table.name,
+                        ", ".join("'%s'" % c.name for c in table_pks),
+                        ", ".join("'%s'" % c.name for c in self.columns),
+                        ", ".join("'%s'" % c.name for c in self.columns)
+                    )
+                )
+            table_pks[:] = []
 
         for c in self.columns:
             c.primary_key = True
+        self.columns.extend(table_pks)
+
+    def _reload(self):
+        """repopulate this :class:`.PrimaryKeyConstraint` based on the current
+        columns marked with primary_key=True in the table.
+
+        Also fires a new event.
+
+        This is basically like putting a whole new
+        :class:`.PrimaryKeyConstraint` object on the parent
+        :class:`.Table` object without actually replacing the object.
+
+        """
+
+        # clear out the columns collection; we will re-populate
+        # based on current primary_key flags
+        self.columns.clear()
+
+        # fire a new event; this will add all existing
+        # primary key columns based on the flag.
+        self._set_parent_with_dispatch(self.table)
 
     def _replace(self, col):
         self.columns.replace(col)
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py
index bd065103e..2f311f7e7 100644
--- a/test/engine/test_reflection.py
+++ b/test/engine/test_reflection.py
@@ -361,6 +361,27 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         self.assert_(isinstance(table.c.col4.type, sa.String))
 
     @testing.provide_metadata
+    def test_override_upgrade_pk_flag(self):
+        meta = self.metadata
+        table = Table(
+            'override_test', meta,
+            Column('col1', sa.Integer),
+            Column('col2', sa.String(20)),
+            Column('col3', sa.Numeric)
+        )
+        table.create()
+
+        meta2 = MetaData(testing.db)
+        table = Table(
+            'override_test', meta2,
+            Column('col1', sa.Integer, primary_key=True),
+            autoload=True)
+
+        eq_(list(table.primary_key), [table.c.col1])
+        eq_(table.c.col1.primary_key, True)
+
+
+    @testing.provide_metadata
     def test_override_pkfk(self):
         """test that you can override columns which contain foreign keys
         to other reflected tables, where the foreign key column is also
diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index 2a52428dd..f933a2494 100644
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -5,7 +5,7 @@ from sqlalchemy.testing import emits_warning
 import pickle
 from sqlalchemy import Integer, String, UniqueConstraint, \
     CheckConstraint, ForeignKey, MetaData, Sequence, \
-    ForeignKeyConstraint, ColumnDefault, Index, event,\
+    ForeignKeyConstraint, PrimaryKeyConstraint, ColumnDefault, Index, event,\
     events, Unicode, types as sqltypes, bindparam, \
     Table, Column
 from sqlalchemy import schema, exc
@@ -842,6 +842,77 @@ class TableTest(fixtures.TestBase, AssertsCompiledSQL):
         )
         is_(t._autoincrement_column, t.c.id)
 
+    def test_pk_args_standalone(self):
+        m = MetaData()
+        t = Table('t', m,
+                    Column('x', Integer, primary_key=True),
+                    PrimaryKeyConstraint(mssql_clustered=True)
+                )
+        eq_(
+            list(t.primary_key), [t.c.x]
+        )
+        eq_(
+            t.primary_key.dialect_kwargs, {"mssql_clustered": True}
+        )
+
+    def test_pk_cols_sets_flags(self):
+        m = MetaData()
+        t = Table('t', m,
+                    Column('x', Integer),
+                    Column('y', Integer),
+                    Column('z', Integer),
+                    PrimaryKeyConstraint('x', 'y')
+                )
+        eq_(t.c.x.primary_key, True)
+        eq_(t.c.y.primary_key, True)
+        eq_(t.c.z.primary_key, False)
+
+    def test_pk_col_mismatch_one(self):
+        m = MetaData()
+        assert_raises_message(
+            exc.SAWarning,
+            "Table 't' specifies columns 'x' as primary_key=True, "
+                "not matching locally specified columns 'q'",
+            Table, 't', m,
+                Column('x', Integer, primary_key=True),
+                Column('q', Integer),
+                PrimaryKeyConstraint('q')
+        )
+
+    def test_pk_col_mismatch_two(self):
+        m = MetaData()
+        assert_raises_message(
+            exc.SAWarning,
+            "Table 't' specifies columns 'a', 'b', 'c' as primary_key=True, "
+                "not matching locally specified columns 'b', 'c'",
+            Table, 't', m,
+           Column('a', Integer, primary_key=True),
+           Column('b', Integer, primary_key=True),
+           Column('c', Integer, primary_key=True),
+           PrimaryKeyConstraint('b', 'c')
+        )
+
+    @testing.emits_warning("Table 't'")
+    def test_pk_col_mismatch_three(self):
+        m = MetaData()
+        t = Table('t', m,
+                Column('x', Integer, primary_key=True),
+                Column('q', Integer),
+                PrimaryKeyConstraint('q')
+        )
+        eq_(list(t.primary_key), [t.c.q])
+
+    @testing.emits_warning("Table 't'")
+    def test_pk_col_mismatch_four(self):
+        m = MetaData()
+        t = Table('t', m,
+           Column('a', Integer, primary_key=True),
+           Column('b', Integer, primary_key=True),
+           Column('c', Integer, primary_key=True),
+           PrimaryKeyConstraint('b', 'c')
+        )
+        eq_(list(t.primary_key), [t.c.b, t.c.c])
+
 class SchemaTypeTest(fixtures.TestBase):
     class MyType(sqltypes.SchemaType, sqltypes.TypeEngine):
         column = None