我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.schema.PrimaryKeyConstraint()。
def _mysql_drop_constraint(element, compiler, **kw): """Redefine SQLAlchemy's drop constraint to raise errors for invalid constraint type.""" constraint = element.element if isinstance(constraint, (schema.ForeignKeyConstraint, schema.PrimaryKeyConstraint, schema.UniqueConstraint) ): return compiler.visit_drop_constraint(element, **kw) elif isinstance(constraint, schema.CheckConstraint): raise NotImplementedError( "MySQL does not support CHECK constraints.") else: raise NotImplementedError( "No generic 'DROP CONSTRAINT' in MySQL - " "please specify constraint type")
def generic_constraint(self, name, table_name, type_, schema=None, **kw): t = self.table(table_name, schema=schema) types = { 'foreignkey': lambda name: sa_schema.ForeignKeyConstraint( [], [], name=name), 'primary': sa_schema.PrimaryKeyConstraint, 'unique': sa_schema.UniqueConstraint, 'check': lambda name: sa_schema.CheckConstraint("", name=name), None: sa_schema.Constraint } try: const = types[type_] except KeyError: raise TypeError("'type' can be one of %s" % ", ".join(sorted(repr(x) for x in types))) else: const = const(name=name) t.append_constraint(const) return const
def add_column(operations, operation): table_name = operation.table_name column = operation.column schema = operation.schema t = operations.schema_obj.table(table_name, column, schema=schema) operations.impl.add_column( table_name, column, schema=schema ) for constraint in t.constraints: if not isinstance(constraint, sa_schema.PrimaryKeyConstraint): operations.impl.add_constraint(constraint) for index in t.indexes: operations.impl.create_index(index)
def drop_constraint(self, const): if not const.name: raise ValueError("Constraint must have a name") try: const = self.named_constraints.pop(const.name) except KeyError: if _is_type_bound(const): # type-bound constraints are only included in the new # table via their type object in any case, so ignore the # drop_constraint() that comes here via the # Operations.implementation_for(alter_column) return raise ValueError("No such constraint: '%s'" % const.name) else: if isinstance(const, PrimaryKeyConstraint): for col in const.columns: self.columns[col.name].primary_key = False
def primary_key_constraint(self, name, table_name, cols, schema=None): m = self.metadata() columns = [sa_schema.Column(n, NULLTYPE) for n in cols] t = sa_schema.Table( table_name, m, *columns, schema=schema) p = sa_schema.PrimaryKeyConstraint( *[t.c[n] for n in cols], name=name) t.append_constraint(p) return p
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 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 visit_column(self, column): """Firebird supports 'DROP col' instead of 'DROP COLUMN col' syntax Drop primary key and unique constraints if dropped column is referencing it.""" if column.primary_key: if column.table.primary_key.columns.contains_column(column): column.table.primary_key.drop() # TODO: recreate primary key if it references more than this column for index in column.table.indexes: # "column in index.columns" causes problems as all # column objects compare equal and return a SQL expression if column.name in [col.name for col in index.columns]: index.drop() # TODO: recreate index if it references more than this column for cons in column.table.constraints: if isinstance(cons,PrimaryKeyConstraint): # will be deleted only when the column its on # is deleted! continue should_drop = column.name in cons.columns if should_drop: self.start_alter_table(column) self.append("DROP CONSTRAINT ") self.append(self.preparer.format_constraint(cons)) self.execute() # TODO: recreate unique constraint if it refenrences more than this column self.start_alter_table(column) self.append('DROP %s' % self.preparer.format_column(column)) self.execute()
def __init__(self, *cols, **kwargs): colnames, table = self._normalize_columns(cols) table = kwargs.pop('table', table) super(PrimaryKeyConstraint, self).__init__(*colnames, **kwargs) if table is not None: self._set_parent(table)
def _primary_key_constraint(self, name, table_name, cols, schema=None): m = self._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 drop_constraint(self, name, table_name, type_=None, schema=None): """Drop a constraint of the given name, typically via DROP CONSTRAINT. :param name: name of the constraint. :param table_name: table name. :param ``type_``: optional, required on MySQL. can be 'foreignkey', 'primary', 'unique', or 'check'. :param schema: Optional schema name to operate within. To control quoting of the schema outside of the default behavior, use the SQLAlchemy construct :class:`~sqlalchemy.sql.elements.quoted_name`. .. versionadded:: 0.7.0 'schema' can now accept a :class:`~sqlalchemy.sql.elements.quoted_name` construct. """ t = self._table(table_name, schema=schema) types = { 'foreignkey': lambda name: sa_schema.ForeignKeyConstraint( [], [], name=name), 'primary': sa_schema.PrimaryKeyConstraint, 'unique': sa_schema.UniqueConstraint, 'check': lambda name: sa_schema.CheckConstraint("", name=name), None: sa_schema.Constraint } try: const = types[type_] except KeyError: raise TypeError("'type' can be one of %s" % ", ".join(sorted(repr(x) for x in types))) const = const(name=name) t.append_constraint(const) self.impl.drop_constraint(const)
