yield op
alembic.op._remove_proxy()
+
+ def _primary_key_constraint(self, name, table_name, cols, schema=None):
+ m = sa_schema.MetaData()
+ columns = [sa_schema.Column(n, NULLTYPE) for n in cols]
+ t1 = sa_schema.Table(table_name, m,
+ *columns,
+ schema=schema)
+ p = sa_schema.PrimaryKeyConstraint(*columns, name=name)
+ t1.append_constraint(p)
+ return p
+
def _foreign_key_constraint(self, name, source, referent,
local_cols, remote_cols,
onupdate=None, ondelete=None,
)
+ def create_primary_key(self, name, table_name, cols, schema=None):
+ """Issue a "create primary key" instruction using the current
+ migration context.
+
+ e.g.::
+
+ from alembic import op
+ op.create_primary_key(
+ "pk_my_table", "my_table",
+ ["id", "version"]
+ )
+
+ This internally generates a :class:`~sqlalchemy.schema.Table` object
+ containing the necessary columns, then generates a new
+ :class:`~sqlalchemy.schema.PrimaryKeyConstraint`
+ object which it then associates with the :class:`~sqlalchemy.schema.Table`.
+ Any event listeners associated with this action will be fired
+ off normally. The :class:`~sqlalchemy.schema.AddConstraint`
+ construct is ultimately used to generate the ALTER statement.
+
+ .. versionadded:: 0.5.0
+
+ :param name: Name of the primary key constraint. The name is necessary
+ so that an ALTER statement can be emitted. For setups that
+ use an automated naming scheme such as that described at
+ `NamingConventions <http://www.sqlalchemy.org/trac/wiki/UsageRecipes/NamingConventions>`_,
+ ``name`` here can be ``None``, as the event listener will
+ apply the name to the constraint object when it is associated
+ with the table.
+ :param table_name: String name of the target table.
+ :param cols: a list of string column names to be applied to the
+ primary key constraint.
+ :param schema: Optional schema name of the table.
+
+ """
+ self.impl.add_constraint(
+ self._primary_key_constraint(name, table_name, cols,
+ schema)
+ )
+
def create_foreign_key(self, name, source, referent, local_cols,
remote_cols, onupdate=None, ondelete=None,
source_schema=None, referent_schema=None):
context = op_fixture()
op.create_foreign_key('fk_test', 't1', 't2',
['foo', 'bar'], ['bat', 'hoho'],
- source_schema='foo2', referent_schema='bar2')
+ source_schema='foo2', referent_schema='bar2')
context.assert_(
"ALTER TABLE foo2.t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
"REFERENCES bar2.t2 (bat, hoho)"
"FOREIGN KEY(foo) REFERENCES t1 (bar)"
)
+def test_add_primary_key_constraint():
+ context = op_fixture()
+ op.create_primary_key("pk_test", "t1", ["foo", "bar"])
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo, bar)"
+ )
+
+def test_add_primary_key_constraint_schema():
+ context = op_fixture()
+ op.create_primary_key("pk_test", "t1", ["foo"], schema="bar")
+ context.assert_(
+ "ALTER TABLE bar.t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo)"
+ )
+
def test_add_check_constraint():
context = op_fixture()
op.create_check_constraint(