mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-06 17:01:07 -04:00
Mike Bayer
d6f11d9030
Added new parameters :paramref:`.AddConstraint.isolate_from_table` and :paramref:`.DropConstraint.isolate_from_table`, defaulting to True, which both document and allow to be controllable the long-standing behavior of these two constructs blocking the given constraint from being included inline within the "CREATE TABLE" sequence, under the assumption that separate add/drop directives were to be used. Fixes: #12382 Change-Id: I53c4170ccb5803f69945ba7aa3d3a143131508eb
1383 lines
43 KiB
Python
1383 lines
43 KiB
Python
from sqlalchemy import CheckConstraint
|
|
from sqlalchemy import Column
|
|
from sqlalchemy import exc
|
|
from sqlalchemy import ForeignKey
|
|
from sqlalchemy import ForeignKeyConstraint
|
|
from sqlalchemy import func
|
|
from sqlalchemy import Index
|
|
from sqlalchemy import Integer
|
|
from sqlalchemy import MetaData
|
|
from sqlalchemy import PrimaryKeyConstraint
|
|
from sqlalchemy import schema
|
|
from sqlalchemy import String
|
|
from sqlalchemy import Table
|
|
from sqlalchemy import testing
|
|
from sqlalchemy import text
|
|
from sqlalchemy import UniqueConstraint
|
|
from sqlalchemy.engine import default
|
|
from sqlalchemy.testing import assert_raises
|
|
from sqlalchemy.testing import assert_raises_message
|
|
from sqlalchemy.testing import AssertsCompiledSQL
|
|
from sqlalchemy.testing import AssertsExecutionResults
|
|
from sqlalchemy.testing import engines
|
|
from sqlalchemy.testing import eq_
|
|
from sqlalchemy.testing import fixtures
|
|
from sqlalchemy.testing.assertions import expect_warnings
|
|
from sqlalchemy.testing.assertsql import AllOf
|
|
from sqlalchemy.testing.assertsql import CompiledSQL
|
|
from sqlalchemy.testing.assertsql import DialectSQL
|
|
from sqlalchemy.testing.assertsql import RegexSQL
|
|
|
|
|
|
class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
|
|
__dialect__ = "default"
|
|
__backend__ = True
|
|
|
|
@testing.provide_metadata
|
|
def test_pk_fk_constraint_create(self):
|
|
metadata = self.metadata
|
|
|
|
Table(
|
|
"employees",
|
|
metadata,
|
|
Column("id", Integer),
|
|
Column("soc", String(40)),
|
|
Column("name", String(30)),
|
|
PrimaryKeyConstraint("id", "soc"),
|
|
)
|
|
Table(
|
|
"elements",
|
|
metadata,
|
|
Column("id", Integer),
|
|
Column("stuff", String(30)),
|
|
Column("emp_id", Integer),
|
|
Column("emp_soc", String(40)),
|
|
PrimaryKeyConstraint("id", name="elements_primkey"),
|
|
ForeignKeyConstraint(
|
|
["emp_id", "emp_soc"], ["employees.id", "employees.soc"]
|
|
),
|
|
)
|
|
self.assert_sql_execution(
|
|
testing.db,
|
|
lambda: metadata.create_all(testing.db, checkfirst=False),
|
|
CompiledSQL(
|
|
"CREATE TABLE employees ("
|
|
"id INTEGER NOT NULL, "
|
|
"soc VARCHAR(40) NOT NULL, "
|
|
"name VARCHAR(30), "
|
|
"PRIMARY KEY (id, soc)"
|
|
")"
|
|
),
|
|
CompiledSQL(
|
|
"CREATE TABLE elements ("
|
|
"id INTEGER NOT NULL, "
|
|
"stuff VARCHAR(30), "
|
|
"emp_id INTEGER, "
|
|
"emp_soc VARCHAR(40), "
|
|
"CONSTRAINT elements_primkey PRIMARY KEY (id), "
|
|
"FOREIGN KEY(emp_id, emp_soc) "
|
|
"REFERENCES employees (id, soc)"
|
|
")"
|
|
),
|
|
)
|
|
|
|
@testing.force_drop_names("a", "b")
|
|
def test_fk_cant_drop_cycled_unnamed(self):
|
|
metadata = MetaData()
|
|
|
|
Table(
|
|
"a",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("bid", Integer),
|
|
ForeignKeyConstraint(["bid"], ["b.id"]),
|
|
)
|
|
Table(
|
|
"b",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("aid", Integer),
|
|
ForeignKeyConstraint(["aid"], ["a.id"]),
|
|
)
|
|
metadata.create_all(testing.db)
|
|
if testing.db.dialect.supports_alter:
|
|
assert_raises_message(
|
|
exc.CircularDependencyError,
|
|
"Can't sort tables for DROP; an unresolvable foreign key "
|
|
"dependency exists between tables: a, b. Please ensure "
|
|
"that the ForeignKey and ForeignKeyConstraint objects "
|
|
"involved in the cycle have names so that they can be "
|
|
"dropped using DROP CONSTRAINT.",
|
|
metadata.drop_all,
|
|
testing.db,
|
|
)
|
|
else:
|
|
with expect_warnings(
|
|
"Can't sort tables for DROP; an unresolvable "
|
|
"foreign key dependency "
|
|
):
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.drop_all(testing.db, checkfirst=False)
|
|
|
|
asserter.assert_(
|
|
AllOf(CompiledSQL("DROP TABLE a"), CompiledSQL("DROP TABLE b"))
|
|
)
|
|
|
|
@testing.provide_metadata
|
|
def test_fk_table_auto_alter_constraint_create(self):
|
|
metadata = self.metadata
|
|
|
|
Table(
|
|
"a",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("bid", Integer),
|
|
ForeignKeyConstraint(["bid"], ["b.id"]),
|
|
)
|
|
Table(
|
|
"b",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("aid", Integer),
|
|
ForeignKeyConstraint(["aid"], ["a.id"], name="bfk"),
|
|
)
|
|
self._assert_cyclic_constraint(
|
|
metadata, auto=True, sqlite_warning=True
|
|
)
|
|
|
|
@testing.provide_metadata
|
|
def test_fk_column_auto_alter_inline_constraint_create(self):
|
|
metadata = self.metadata
|
|
|
|
Table(
|
|
"a",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("bid", Integer, ForeignKey("b.id")),
|
|
)
|
|
Table(
|
|
"b",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("aid", Integer, ForeignKey("a.id", name="bfk")),
|
|
)
|
|
self._assert_cyclic_constraint(
|
|
metadata, auto=True, sqlite_warning=True
|
|
)
|
|
|
|
@testing.provide_metadata
|
|
def test_fk_column_use_alter_inline_constraint_create(self):
|
|
metadata = self.metadata
|
|
|
|
Table(
|
|
"a",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("bid", Integer, ForeignKey("b.id")),
|
|
)
|
|
Table(
|
|
"b",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column(
|
|
"aid", Integer, ForeignKey("a.id", name="bfk", use_alter=True)
|
|
),
|
|
)
|
|
self._assert_cyclic_constraint(metadata, auto=False)
|
|
|
|
@testing.provide_metadata
|
|
def test_fk_table_use_alter_constraint_create(self):
|
|
metadata = self.metadata
|
|
|
|
Table(
|
|
"a",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("bid", Integer),
|
|
ForeignKeyConstraint(["bid"], ["b.id"]),
|
|
)
|
|
Table(
|
|
"b",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("aid", Integer),
|
|
ForeignKeyConstraint(
|
|
["aid"], ["a.id"], use_alter=True, name="bfk"
|
|
),
|
|
)
|
|
self._assert_cyclic_constraint(metadata)
|
|
|
|
@testing.provide_metadata
|
|
def test_fk_column_use_alter_constraint_create(self):
|
|
metadata = self.metadata
|
|
|
|
Table(
|
|
"a",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("bid", Integer, ForeignKey("b.id")),
|
|
)
|
|
Table(
|
|
"b",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column(
|
|
"aid", Integer, ForeignKey("a.id", use_alter=True, name="bfk")
|
|
),
|
|
)
|
|
self._assert_cyclic_constraint(metadata, auto=False)
|
|
|
|
def _assert_cyclic_constraint(
|
|
self, metadata, auto=False, sqlite_warning=False
|
|
):
|
|
if testing.db.dialect.supports_alter:
|
|
self._assert_cyclic_constraint_supports_alter(metadata, auto=auto)
|
|
else:
|
|
self._assert_cyclic_constraint_no_alter(
|
|
metadata, auto=auto, sqlite_warning=sqlite_warning
|
|
)
|
|
|
|
def _assert_cyclic_constraint_supports_alter(self, metadata, auto=False):
|
|
table_assertions = []
|
|
if auto:
|
|
table_assertions = [
|
|
CompiledSQL(
|
|
"CREATE TABLE b ("
|
|
"id INTEGER NOT NULL, "
|
|
"aid INTEGER, "
|
|
"PRIMARY KEY (id)"
|
|
")"
|
|
),
|
|
CompiledSQL(
|
|
"CREATE TABLE a ("
|
|
"id INTEGER NOT NULL, "
|
|
"bid INTEGER, "
|
|
"PRIMARY KEY (id)"
|
|
")"
|
|
),
|
|
]
|
|
else:
|
|
table_assertions = [
|
|
CompiledSQL(
|
|
"CREATE TABLE b ("
|
|
"id INTEGER NOT NULL, "
|
|
"aid INTEGER, "
|
|
"PRIMARY KEY (id)"
|
|
")"
|
|
),
|
|
CompiledSQL(
|
|
"CREATE TABLE a ("
|
|
"id INTEGER NOT NULL, "
|
|
"bid INTEGER, "
|
|
"PRIMARY KEY (id), "
|
|
"FOREIGN KEY(bid) REFERENCES b (id)"
|
|
")"
|
|
),
|
|
]
|
|
|
|
assertions = [AllOf(*table_assertions)]
|
|
fk_assertions = []
|
|
fk_assertions.append(
|
|
CompiledSQL(
|
|
"ALTER TABLE b ADD CONSTRAINT bfk "
|
|
"FOREIGN KEY(aid) REFERENCES a (id)"
|
|
)
|
|
)
|
|
if auto:
|
|
fk_assertions.append(
|
|
CompiledSQL(
|
|
"ALTER TABLE a ADD FOREIGN KEY(bid) REFERENCES b (id)"
|
|
)
|
|
)
|
|
assertions.append(AllOf(*fk_assertions))
|
|
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.create_all(testing.db, checkfirst=False)
|
|
asserter.assert_(*assertions)
|
|
|
|
assertions = [
|
|
CompiledSQL("ALTER TABLE b DROP CONSTRAINT bfk"),
|
|
CompiledSQL("DROP TABLE a"),
|
|
CompiledSQL("DROP TABLE b"),
|
|
]
|
|
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.drop_all(testing.db, checkfirst=False),
|
|
asserter.assert_(*assertions)
|
|
|
|
def _assert_cyclic_constraint_no_alter(
|
|
self, metadata, auto=False, sqlite_warning=False
|
|
):
|
|
table_assertions = []
|
|
if auto:
|
|
table_assertions.append(
|
|
DialectSQL(
|
|
"CREATE TABLE b ("
|
|
"id INTEGER NOT NULL, "
|
|
"aid INTEGER, "
|
|
"PRIMARY KEY (id), "
|
|
"CONSTRAINT bfk FOREIGN KEY(aid) REFERENCES a (id)"
|
|
")"
|
|
)
|
|
)
|
|
table_assertions.append(
|
|
DialectSQL(
|
|
"CREATE TABLE a ("
|
|
"id INTEGER NOT NULL, "
|
|
"bid INTEGER, "
|
|
"PRIMARY KEY (id), "
|
|
"FOREIGN KEY(bid) REFERENCES b (id)"
|
|
")"
|
|
)
|
|
)
|
|
else:
|
|
table_assertions.append(
|
|
DialectSQL(
|
|
"CREATE TABLE b ("
|
|
"id INTEGER NOT NULL, "
|
|
"aid INTEGER, "
|
|
"PRIMARY KEY (id), "
|
|
"CONSTRAINT bfk FOREIGN KEY(aid) REFERENCES a (id)"
|
|
")"
|
|
)
|
|
)
|
|
|
|
table_assertions.append(
|
|
DialectSQL(
|
|
"CREATE TABLE a ("
|
|
"id INTEGER NOT NULL, "
|
|
"bid INTEGER, "
|
|
"PRIMARY KEY (id), "
|
|
"FOREIGN KEY(bid) REFERENCES b (id)"
|
|
")"
|
|
)
|
|
)
|
|
|
|
assertions = [AllOf(*table_assertions)]
|
|
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.create_all(testing.db, checkfirst=False)
|
|
asserter.assert_(*assertions)
|
|
|
|
assertions = [
|
|
AllOf(CompiledSQL("DROP TABLE a"), CompiledSQL("DROP TABLE b"))
|
|
]
|
|
|
|
if sqlite_warning:
|
|
with expect_warnings("Can't sort tables for DROP; "):
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.drop_all(testing.db, checkfirst=False),
|
|
else:
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.drop_all(testing.db, checkfirst=False),
|
|
asserter.assert_(*assertions)
|
|
|
|
@testing.force_drop_names("a", "b")
|
|
def test_cycle_unnamed_fks(self):
|
|
metadata = MetaData()
|
|
|
|
Table(
|
|
"a",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("bid", Integer, ForeignKey("b.id")),
|
|
)
|
|
|
|
Table(
|
|
"b",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("aid", Integer, ForeignKey("a.id")),
|
|
)
|
|
|
|
assertions = [
|
|
AllOf(
|
|
CompiledSQL(
|
|
"CREATE TABLE b ("
|
|
"id INTEGER NOT NULL, "
|
|
"aid INTEGER, "
|
|
"PRIMARY KEY (id)"
|
|
")"
|
|
),
|
|
CompiledSQL(
|
|
"CREATE TABLE a ("
|
|
"id INTEGER NOT NULL, "
|
|
"bid INTEGER, "
|
|
"PRIMARY KEY (id)"
|
|
")"
|
|
),
|
|
),
|
|
AllOf(
|
|
CompiledSQL(
|
|
"ALTER TABLE b ADD FOREIGN KEY(aid) REFERENCES a (id)"
|
|
),
|
|
CompiledSQL(
|
|
"ALTER TABLE a ADD FOREIGN KEY(bid) REFERENCES b (id)"
|
|
),
|
|
),
|
|
]
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.create_all(testing.db, checkfirst=False)
|
|
|
|
if testing.db.dialect.supports_alter:
|
|
asserter.assert_(*assertions)
|
|
|
|
assert_raises_message(
|
|
exc.CircularDependencyError,
|
|
"Can't sort tables for DROP; an unresolvable foreign key "
|
|
"dependency exists between tables: a, b. "
|
|
"Please ensure that the "
|
|
"ForeignKey and ForeignKeyConstraint objects involved in the "
|
|
"cycle have names so that they can be dropped using "
|
|
"DROP CONSTRAINT.",
|
|
metadata.drop_all,
|
|
testing.db,
|
|
checkfirst=False,
|
|
)
|
|
else:
|
|
with expect_warnings(
|
|
"Can't sort tables for DROP; an unresolvable "
|
|
"foreign key dependency exists between tables"
|
|
):
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.drop_all(testing.db, checkfirst=False)
|
|
|
|
asserter.assert_(
|
|
AllOf(CompiledSQL("DROP TABLE b"), CompiledSQL("DROP TABLE a"))
|
|
)
|
|
|
|
@testing.force_drop_names("a", "b")
|
|
def test_cycle_named_fks(self):
|
|
metadata = MetaData()
|
|
|
|
Table(
|
|
"a",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("bid", Integer, ForeignKey("b.id")),
|
|
)
|
|
|
|
Table(
|
|
"b",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column(
|
|
"aid",
|
|
Integer,
|
|
ForeignKey("a.id", use_alter=True, name="aidfk"),
|
|
),
|
|
)
|
|
|
|
assertions = [
|
|
AllOf(
|
|
CompiledSQL(
|
|
"CREATE TABLE b ("
|
|
"id INTEGER NOT NULL, "
|
|
"aid INTEGER, "
|
|
"PRIMARY KEY (id)"
|
|
")"
|
|
),
|
|
CompiledSQL(
|
|
"CREATE TABLE a ("
|
|
"id INTEGER NOT NULL, "
|
|
"bid INTEGER, "
|
|
"PRIMARY KEY (id), "
|
|
"FOREIGN KEY(bid) REFERENCES b (id)"
|
|
")"
|
|
),
|
|
),
|
|
CompiledSQL(
|
|
"ALTER TABLE b ADD CONSTRAINT aidfk "
|
|
"FOREIGN KEY(aid) REFERENCES a (id)"
|
|
),
|
|
]
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.create_all(testing.db, checkfirst=False)
|
|
|
|
if testing.db.dialect.supports_alter:
|
|
asserter.assert_(*assertions)
|
|
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.drop_all(testing.db, checkfirst=False)
|
|
|
|
asserter.assert_(
|
|
CompiledSQL("ALTER TABLE b DROP CONSTRAINT aidfk"),
|
|
AllOf(
|
|
CompiledSQL("DROP TABLE b"), CompiledSQL("DROP TABLE a")
|
|
),
|
|
)
|
|
else:
|
|
with self.sql_execution_asserter() as asserter:
|
|
metadata.drop_all(testing.db, checkfirst=False)
|
|
|
|
asserter.assert_(
|
|
AllOf(CompiledSQL("DROP TABLE b"), CompiledSQL("DROP TABLE a"))
|
|
)
|
|
|
|
@testing.requires.check_constraints
|
|
@testing.provide_metadata
|
|
def test_check_constraint_create(self):
|
|
metadata = self.metadata
|
|
|
|
Table(
|
|
"foo",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("x", Integer),
|
|
Column("y", Integer),
|
|
CheckConstraint("x>y"),
|
|
)
|
|
Table(
|
|
"bar",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("x", Integer, CheckConstraint("x>7")),
|
|
Column("z", Integer),
|
|
)
|
|
|
|
self.assert_sql_execution(
|
|
testing.db,
|
|
lambda: metadata.create_all(testing.db, checkfirst=False),
|
|
AllOf(
|
|
CompiledSQL(
|
|
"CREATE TABLE foo ("
|
|
"id INTEGER NOT NULL, "
|
|
"x INTEGER, "
|
|
"y INTEGER, "
|
|
"PRIMARY KEY (id), "
|
|
"CHECK (x>y)"
|
|
")"
|
|
),
|
|
CompiledSQL(
|
|
"CREATE TABLE bar ("
|
|
"id INTEGER NOT NULL, "
|
|
"x INTEGER CHECK (x>7), "
|
|
"z INTEGER, "
|
|
"PRIMARY KEY (id)"
|
|
")"
|
|
),
|
|
),
|
|
)
|
|
|
|
@testing.provide_metadata
|
|
def test_unique_constraint_create(self):
|
|
metadata = self.metadata
|
|
|
|
Table(
|
|
"foo",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("value", String(30), unique=True),
|
|
)
|
|
Table(
|
|
"bar",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("value", String(30)),
|
|
Column("value2", String(30)),
|
|
UniqueConstraint("value", "value2", name="uix1"),
|
|
)
|
|
|
|
self.assert_sql_execution(
|
|
testing.db,
|
|
lambda: metadata.create_all(testing.db, checkfirst=False),
|
|
AllOf(
|
|
CompiledSQL(
|
|
"CREATE TABLE foo ("
|
|
"id INTEGER NOT NULL, "
|
|
"value VARCHAR(30), "
|
|
"PRIMARY KEY (id), "
|
|
"UNIQUE (value)"
|
|
")"
|
|
),
|
|
CompiledSQL(
|
|
"CREATE TABLE bar ("
|
|
"id INTEGER NOT NULL, "
|
|
"value VARCHAR(30), "
|
|
"value2 VARCHAR(30), "
|
|
"PRIMARY KEY (id), "
|
|
"CONSTRAINT uix1 UNIQUE (value, value2)"
|
|
")"
|
|
),
|
|
),
|
|
)
|
|
|
|
@testing.provide_metadata
|
|
def test_index_create(self):
|
|
metadata = self.metadata
|
|
|
|
employees = Table(
|
|
"employees",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("first_name", String(30)),
|
|
Column("last_name", String(30)),
|
|
Column("email_address", String(30)),
|
|
)
|
|
|
|
i = Index(
|
|
"employee_name_index",
|
|
employees.c.last_name,
|
|
employees.c.first_name,
|
|
)
|
|
assert i in employees.indexes
|
|
|
|
i2 = Index(
|
|
"employee_email_index", employees.c.email_address, unique=True
|
|
)
|
|
assert i2 in employees.indexes
|
|
|
|
self.assert_sql_execution(
|
|
testing.db,
|
|
lambda: metadata.create_all(testing.db, checkfirst=False),
|
|
RegexSQL("^CREATE TABLE"),
|
|
AllOf(
|
|
CompiledSQL(
|
|
"CREATE INDEX employee_name_index ON "
|
|
"employees (last_name, first_name)",
|
|
[],
|
|
),
|
|
CompiledSQL(
|
|
"CREATE UNIQUE INDEX employee_email_index ON "
|
|
"employees (email_address)",
|
|
[],
|
|
),
|
|
),
|
|
)
|
|
|
|
@testing.provide_metadata
|
|
def test_index_create_camelcase(self):
|
|
"""test that mixed-case index identifiers are legal"""
|
|
|
|
metadata = self.metadata
|
|
|
|
employees = Table(
|
|
"companyEmployees",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("firstName", String(30)),
|
|
Column("lastName", String(30)),
|
|
Column("emailAddress", String(30)),
|
|
)
|
|
|
|
Index("employeeNameIndex", employees.c.lastName, employees.c.firstName)
|
|
|
|
Index("employeeEmailIndex", employees.c.emailAddress, unique=True)
|
|
|
|
self.assert_sql_execution(
|
|
testing.db,
|
|
lambda: metadata.create_all(testing.db, checkfirst=False),
|
|
RegexSQL("^CREATE TABLE"),
|
|
AllOf(
|
|
CompiledSQL(
|
|
'CREATE INDEX "employeeNameIndex" ON '
|
|
'"companyEmployees" ("lastName", "firstName")',
|
|
[],
|
|
),
|
|
CompiledSQL(
|
|
'CREATE UNIQUE INDEX "employeeEmailIndex" ON '
|
|
'"companyEmployees" ("emailAddress")',
|
|
[],
|
|
),
|
|
),
|
|
)
|
|
|
|
@testing.provide_metadata
|
|
def test_index_create_inline(self):
|
|
# test an index create using index=True, unique=True
|
|
|
|
metadata = self.metadata
|
|
|
|
events = Table(
|
|
"events",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("name", String(30), index=True, unique=True),
|
|
Column("location", String(30), index=True),
|
|
Column("sport", String(30)),
|
|
Column("announcer", String(30)),
|
|
Column("winner", String(30)),
|
|
)
|
|
|
|
Index(
|
|
"sport_announcer", events.c.sport, events.c.announcer, unique=True
|
|
)
|
|
Index("idx_winners", events.c.winner)
|
|
|
|
eq_(
|
|
{ix.name for ix in events.indexes},
|
|
{
|
|
"ix_events_name",
|
|
"ix_events_location",
|
|
"sport_announcer",
|
|
"idx_winners",
|
|
},
|
|
)
|
|
|
|
self.assert_sql_execution(
|
|
testing.db,
|
|
lambda: events.create(testing.db),
|
|
RegexSQL("^CREATE TABLE events"),
|
|
AllOf(
|
|
CompiledSQL(
|
|
"CREATE UNIQUE INDEX ix_events_name ON events (name)"
|
|
),
|
|
CompiledSQL(
|
|
"CREATE INDEX ix_events_location ON events (location)"
|
|
),
|
|
CompiledSQL(
|
|
"CREATE UNIQUE INDEX sport_announcer ON events "
|
|
"(sport, announcer)"
|
|
),
|
|
CompiledSQL("CREATE INDEX idx_winners ON events (winner)"),
|
|
),
|
|
)
|
|
|
|
@testing.provide_metadata
|
|
def test_index_functional_create(self):
|
|
metadata = self.metadata
|
|
|
|
t = Table(
|
|
"sometable",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("data", String(50)),
|
|
)
|
|
Index("myindex", t.c.data.desc())
|
|
self.assert_sql_execution(
|
|
testing.db,
|
|
lambda: t.create(testing.db),
|
|
CompiledSQL(
|
|
"CREATE TABLE sometable (id INTEGER NOT NULL, "
|
|
"data VARCHAR(50), PRIMARY KEY (id))"
|
|
),
|
|
CompiledSQL("CREATE INDEX myindex ON sometable (data DESC)"),
|
|
)
|
|
|
|
|
|
class ConstraintCompilationTest(fixtures.TestBase, AssertsCompiledSQL):
|
|
__dialect__ = "default"
|
|
|
|
def test_create_index_plain(self):
|
|
t = Table("t", MetaData(), Column("x", Integer))
|
|
i = Index("xyz", t.c.x)
|
|
self.assert_compile(schema.CreateIndex(i), "CREATE INDEX xyz ON t (x)")
|
|
|
|
def test_create_index_if_not_exists(self):
|
|
t = Table("t", MetaData(), Column("x", Integer))
|
|
i = Index("xyz", t.c.x)
|
|
self.assert_compile(
|
|
schema.CreateIndex(i, if_not_exists=True),
|
|
"CREATE INDEX IF NOT EXISTS xyz ON t (x)",
|
|
)
|
|
|
|
def test_drop_index_plain_unattached(self):
|
|
self.assert_compile(
|
|
schema.DropIndex(Index(name="xyz")), "DROP INDEX xyz"
|
|
)
|
|
|
|
def test_drop_index_plain(self):
|
|
self.assert_compile(
|
|
schema.DropIndex(Index(name="xyz")), "DROP INDEX xyz"
|
|
)
|
|
|
|
def test_drop_index_if_exists(self):
|
|
self.assert_compile(
|
|
schema.DropIndex(Index(name="xyz"), if_exists=True),
|
|
"DROP INDEX IF EXISTS xyz",
|
|
)
|
|
|
|
def test_create_index_schema(self):
|
|
t = Table("t", MetaData(), Column("x", Integer), schema="foo")
|
|
i = Index("xyz", t.c.x)
|
|
self.assert_compile(
|
|
schema.CreateIndex(i), "CREATE INDEX xyz ON foo.t (x)"
|
|
)
|
|
|
|
def test_drop_index_schema(self):
|
|
t = Table("t", MetaData(), Column("x", Integer), schema="foo")
|
|
i = Index("xyz", t.c.x)
|
|
self.assert_compile(schema.DropIndex(i), "DROP INDEX foo.xyz")
|
|
|
|
def test_too_long_index_name(self):
|
|
dialect = testing.db.dialect.__class__()
|
|
|
|
for max_ident, max_index in [(22, None), (256, 22)]:
|
|
dialect.max_identifier_length = max_ident
|
|
dialect.max_index_name_length = max_index
|
|
|
|
for tname, cname, exp in [
|
|
("sometable", "this_name_is_too_long", "ix_sometable_t_09aa"),
|
|
("sometable", "this_name_alsois_long", "ix_sometable_t_3cf1"),
|
|
]:
|
|
t1 = Table(
|
|
tname, MetaData(), Column(cname, Integer, index=True)
|
|
)
|
|
ix1 = list(t1.indexes)[0]
|
|
|
|
self.assert_compile(
|
|
schema.CreateIndex(ix1),
|
|
"CREATE INDEX %s ON %s (%s)" % (exp, tname, cname),
|
|
dialect=dialect,
|
|
)
|
|
|
|
dialect.max_identifier_length = 22
|
|
dialect.max_index_name_length = None
|
|
|
|
t1 = Table("t", MetaData(), Column("c", Integer))
|
|
assert_raises(
|
|
exc.IdentifierError,
|
|
schema.CreateIndex(
|
|
Index(
|
|
"this_other_name_is_too_long_for_what_were_doing", t1.c.c
|
|
)
|
|
).compile,
|
|
dialect=dialect,
|
|
)
|
|
|
|
def test_functional_index_w_string_cols_combo(self):
|
|
metadata = MetaData()
|
|
x = Table(
|
|
"x",
|
|
metadata,
|
|
Column("q", String(50)),
|
|
Column("p", Integer),
|
|
Column("z", Integer),
|
|
)
|
|
|
|
for idx, ddl in [
|
|
(
|
|
Index("y", func.lower(x.c.q), "p", x.c.z),
|
|
"CREATE INDEX y ON x (lower(q), p, z)",
|
|
),
|
|
(
|
|
Index("y", "p", func.lower(x.c.q), "z"),
|
|
"CREATE INDEX y ON x (p, lower(q), z)",
|
|
),
|
|
(
|
|
Index("y", "p", "z", func.lower(x.c.q)),
|
|
"CREATE INDEX y ON x (p, z, lower(q))",
|
|
),
|
|
(
|
|
Index("y", func.foo("foob"), x.c.p, "z"),
|
|
"CREATE INDEX y ON x (foo('foob'), p, z)",
|
|
),
|
|
(
|
|
Index("y", x.c.p, func.foo("foob"), "z"),
|
|
"CREATE INDEX y ON x (p, foo('foob'), z)",
|
|
),
|
|
(
|
|
Index("y", func.foo("foob"), "p", "z"),
|
|
"CREATE INDEX y ON x (foo('foob'), p, z)",
|
|
),
|
|
]:
|
|
x.append_constraint(idx)
|
|
self.assert_compile(schema.CreateIndex(idx), ddl)
|
|
|
|
x.to_metadata(MetaData())
|
|
|
|
def test_index_against_text_separate(self):
|
|
metadata = MetaData()
|
|
idx = Index("y", text("some_function(q)"))
|
|
t = Table("x", metadata, Column("q", String(50)))
|
|
t.append_constraint(idx)
|
|
self.assert_compile(
|
|
schema.CreateIndex(idx), "CREATE INDEX y ON x (some_function(q))"
|
|
)
|
|
|
|
def test_index_against_text_inline(self):
|
|
metadata = MetaData()
|
|
idx = Index("y", text("some_function(q)"))
|
|
Table("x", metadata, Column("q", String(50)), idx)
|
|
|
|
self.assert_compile(
|
|
schema.CreateIndex(idx), "CREATE INDEX y ON x (some_function(q))"
|
|
)
|
|
|
|
def test_index_declaration_inline(self):
|
|
metadata = MetaData()
|
|
|
|
t1 = Table(
|
|
"t1",
|
|
metadata,
|
|
Column("x", Integer),
|
|
Column("y", Integer),
|
|
Index("foo", "x", "y"),
|
|
)
|
|
self.assert_compile(
|
|
schema.CreateIndex(list(t1.indexes)[0]),
|
|
"CREATE INDEX foo ON t1 (x, y)",
|
|
)
|
|
|
|
def _test_deferrable(self, constraint_factory):
|
|
dialect = default.DefaultDialect()
|
|
|
|
t = Table(
|
|
"tbl",
|
|
MetaData(),
|
|
Column("a", Integer),
|
|
Column("b", Integer),
|
|
constraint_factory(deferrable=True),
|
|
)
|
|
|
|
sql = str(schema.CreateTable(t).compile(dialect=dialect))
|
|
assert "DEFERRABLE" in sql, sql
|
|
assert "NOT DEFERRABLE" not in sql, sql
|
|
|
|
t = Table(
|
|
"tbl",
|
|
MetaData(),
|
|
Column("a", Integer),
|
|
Column("b", Integer),
|
|
constraint_factory(deferrable=False),
|
|
)
|
|
|
|
sql = str(schema.CreateTable(t).compile(dialect=dialect))
|
|
assert "NOT DEFERRABLE" in sql
|
|
|
|
t = Table(
|
|
"tbl",
|
|
MetaData(),
|
|
Column("a", Integer),
|
|
Column("b", Integer),
|
|
constraint_factory(deferrable=True, initially="IMMEDIATE"),
|
|
)
|
|
sql = str(schema.CreateTable(t).compile(dialect=dialect))
|
|
assert "NOT DEFERRABLE" not in sql
|
|
assert "INITIALLY IMMEDIATE" in sql
|
|
|
|
t = Table(
|
|
"tbl",
|
|
MetaData(),
|
|
Column("a", Integer),
|
|
Column("b", Integer),
|
|
constraint_factory(deferrable=True, initially="DEFERRED"),
|
|
)
|
|
sql = str(schema.CreateTable(t).compile(dialect=dialect))
|
|
|
|
assert "NOT DEFERRABLE" not in sql
|
|
assert "INITIALLY DEFERRED" in sql
|
|
|
|
def test_column_level_ck_name(self):
|
|
t = Table(
|
|
"tbl",
|
|
MetaData(),
|
|
Column(
|
|
"a",
|
|
Integer,
|
|
CheckConstraint("a > 5", name="ck_a_greater_five"),
|
|
),
|
|
)
|
|
self.assert_compile(
|
|
schema.CreateTable(t),
|
|
"CREATE TABLE tbl (a INTEGER CONSTRAINT "
|
|
"ck_a_greater_five CHECK (a > 5))",
|
|
)
|
|
|
|
def test_deferrable_pk(self):
|
|
def factory(**kw):
|
|
return PrimaryKeyConstraint("a", **kw)
|
|
|
|
self._test_deferrable(factory)
|
|
|
|
def test_deferrable_table_fk(self):
|
|
def factory(**kw):
|
|
return ForeignKeyConstraint(["b"], ["tbl.a"], **kw)
|
|
|
|
self._test_deferrable(factory)
|
|
|
|
def test_deferrable_column_fk(self):
|
|
t = Table(
|
|
"tbl",
|
|
MetaData(),
|
|
Column("a", Integer),
|
|
Column(
|
|
"b",
|
|
Integer,
|
|
ForeignKey("tbl.a", deferrable=True, initially="DEFERRED"),
|
|
),
|
|
)
|
|
|
|
self.assert_compile(
|
|
schema.CreateTable(t),
|
|
"CREATE TABLE tbl (a INTEGER, b INTEGER, "
|
|
"FOREIGN KEY(b) REFERENCES tbl "
|
|
"(a) DEFERRABLE INITIALLY DEFERRED)",
|
|
)
|
|
|
|
def test_fk_match_clause(self):
|
|
t = Table(
|
|
"tbl",
|
|
MetaData(),
|
|
Column("a", Integer),
|
|
Column("b", Integer, ForeignKey("tbl.a", match="SIMPLE")),
|
|
)
|
|
|
|
self.assert_compile(
|
|
schema.CreateTable(t),
|
|
"CREATE TABLE tbl (a INTEGER, b INTEGER, "
|
|
"FOREIGN KEY(b) REFERENCES tbl "
|
|
"(a) MATCH SIMPLE)",
|
|
)
|
|
|
|
self.assert_compile(
|
|
schema.AddConstraint(list(t.foreign_keys)[0].constraint),
|
|
"ALTER TABLE tbl ADD FOREIGN KEY(b) "
|
|
"REFERENCES tbl (a) MATCH SIMPLE",
|
|
)
|
|
|
|
def test_create_table_omit_fks(self):
|
|
fkcs = [
|
|
ForeignKeyConstraint(["a"], ["remote.id"], name="foo"),
|
|
ForeignKeyConstraint(["b"], ["remote.id"], name="bar"),
|
|
ForeignKeyConstraint(["c"], ["remote.id"], name="bat"),
|
|
]
|
|
m = MetaData()
|
|
t = Table(
|
|
"t",
|
|
m,
|
|
Column("a", Integer),
|
|
Column("b", Integer),
|
|
Column("c", Integer),
|
|
*fkcs,
|
|
)
|
|
Table("remote", m, Column("id", Integer, primary_key=True))
|
|
|
|
self.assert_compile(
|
|
schema.CreateTable(t, include_foreign_key_constraints=[]),
|
|
"CREATE TABLE t (a INTEGER, b INTEGER, c INTEGER)",
|
|
)
|
|
self.assert_compile(
|
|
schema.CreateTable(t, include_foreign_key_constraints=fkcs[0:2]),
|
|
"CREATE TABLE t (a INTEGER, b INTEGER, c INTEGER, "
|
|
"CONSTRAINT foo FOREIGN KEY(a) REFERENCES remote (id), "
|
|
"CONSTRAINT bar FOREIGN KEY(b) REFERENCES remote (id))",
|
|
)
|
|
|
|
def test_deferrable_unique(self):
|
|
def factory(**kw):
|
|
return UniqueConstraint("b", **kw)
|
|
|
|
self._test_deferrable(factory)
|
|
|
|
def test_deferrable_table_check(self):
|
|
def factory(**kw):
|
|
return CheckConstraint("a < b", **kw)
|
|
|
|
self._test_deferrable(factory)
|
|
|
|
def test_multiple(self):
|
|
m = MetaData()
|
|
Table(
|
|
"foo",
|
|
m,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("bar", Integer, primary_key=True),
|
|
)
|
|
tb = Table(
|
|
"some_table",
|
|
m,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("foo_id", Integer, ForeignKey("foo.id")),
|
|
Column("foo_bar", Integer, ForeignKey("foo.bar")),
|
|
)
|
|
self.assert_compile(
|
|
schema.CreateTable(tb),
|
|
"CREATE TABLE some_table ("
|
|
"id INTEGER NOT NULL, "
|
|
"foo_id INTEGER, "
|
|
"foo_bar INTEGER, "
|
|
"PRIMARY KEY (id), "
|
|
"FOREIGN KEY(foo_id) REFERENCES foo (id), "
|
|
"FOREIGN KEY(foo_bar) REFERENCES foo (bar))",
|
|
)
|
|
|
|
def test_empty_pkc(self):
|
|
# test that an empty primary key is ignored
|
|
metadata = MetaData()
|
|
tbl = Table(
|
|
"test",
|
|
metadata,
|
|
Column("x", Integer, autoincrement=False),
|
|
Column("y", Integer, autoincrement=False),
|
|
PrimaryKeyConstraint(),
|
|
)
|
|
self.assert_compile(
|
|
schema.CreateTable(tbl), "CREATE TABLE test (x INTEGER, y INTEGER)"
|
|
)
|
|
|
|
def test_empty_uc(self):
|
|
# test that an empty constraint is ignored
|
|
metadata = MetaData()
|
|
tbl = Table(
|
|
"test",
|
|
metadata,
|
|
Column("x", Integer, autoincrement=False),
|
|
Column("y", Integer, autoincrement=False),
|
|
UniqueConstraint(),
|
|
)
|
|
self.assert_compile(
|
|
schema.CreateTable(tbl), "CREATE TABLE test (x INTEGER, y INTEGER)"
|
|
)
|
|
|
|
def test_deferrable_column_check(self):
|
|
t = Table(
|
|
"tbl",
|
|
MetaData(),
|
|
Column("a", Integer),
|
|
Column(
|
|
"b",
|
|
Integer,
|
|
CheckConstraint(
|
|
"a < b", deferrable=True, initially="DEFERRED"
|
|
),
|
|
),
|
|
)
|
|
|
|
self.assert_compile(
|
|
schema.CreateTable(t),
|
|
"CREATE TABLE tbl (a INTEGER, b INTEGER CHECK (a < b) "
|
|
"DEFERRABLE INITIALLY DEFERRED)",
|
|
)
|
|
|
|
def test_use_alter(self):
|
|
m = MetaData()
|
|
Table("t", m, Column("a", Integer))
|
|
|
|
Table(
|
|
"t2",
|
|
m,
|
|
Column(
|
|
"a", Integer, ForeignKey("t.a", use_alter=True, name="fk_ta")
|
|
),
|
|
Column("b", Integer, ForeignKey("t.a", name="fk_tb")),
|
|
)
|
|
|
|
e = engines.mock_engine(dialect_name="postgresql")
|
|
m.create_all(e)
|
|
m.drop_all(e)
|
|
|
|
e.assert_sql(
|
|
[
|
|
"CREATE TABLE t (a INTEGER)",
|
|
"CREATE TABLE t2 (a INTEGER, b INTEGER, CONSTRAINT fk_tb "
|
|
"FOREIGN KEY(b) REFERENCES t (a))",
|
|
"ALTER TABLE t2 "
|
|
"ADD CONSTRAINT fk_ta FOREIGN KEY(a) REFERENCES t (a)",
|
|
"ALTER TABLE t2 DROP CONSTRAINT fk_ta",
|
|
"DROP TABLE t2",
|
|
"DROP TABLE t",
|
|
]
|
|
)
|
|
|
|
def _constraint_create_fixture(self):
|
|
m = MetaData()
|
|
|
|
t = Table("tbl", m, Column("a", Integer), Column("b", Integer))
|
|
|
|
t2 = Table("t2", m, Column("a", Integer), Column("b", Integer))
|
|
|
|
return t, t2
|
|
|
|
def test_render_ck_constraint_inline(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
CheckConstraint(
|
|
"a < b",
|
|
name="my_test_constraint",
|
|
deferrable=True,
|
|
initially="DEFERRED",
|
|
table=t,
|
|
)
|
|
|
|
# before we create an AddConstraint,
|
|
# the CONSTRAINT comes out inline
|
|
self.assert_compile(
|
|
schema.CreateTable(t),
|
|
"CREATE TABLE tbl ("
|
|
"a INTEGER, "
|
|
"b INTEGER, "
|
|
"CONSTRAINT my_test_constraint CHECK (a < b) "
|
|
"DEFERRABLE INITIALLY DEFERRED"
|
|
")",
|
|
)
|
|
|
|
def test_render_ck_constraint_external(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = CheckConstraint(
|
|
"a < b",
|
|
name="my_test_constraint",
|
|
deferrable=True,
|
|
initially="DEFERRED",
|
|
table=t,
|
|
)
|
|
|
|
self.assert_compile(
|
|
schema.AddConstraint(constraint),
|
|
"ALTER TABLE tbl ADD CONSTRAINT my_test_constraint "
|
|
"CHECK (a < b) DEFERRABLE INITIALLY DEFERRED",
|
|
)
|
|
|
|
@testing.variation("isolate", [True, False])
|
|
@testing.variation("type_", ["add", "drop"])
|
|
def test_external_ck_constraint_cancels_internal(
|
|
self, isolate: testing.Variation, type_: testing.Variation
|
|
):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = CheckConstraint(
|
|
"a < b",
|
|
name="my_test_constraint",
|
|
deferrable=True,
|
|
initially="DEFERRED",
|
|
table=t,
|
|
)
|
|
|
|
if type_.add:
|
|
cls = schema.AddConstraint
|
|
elif type_.drop:
|
|
cls = schema.DropConstraint
|
|
else:
|
|
type_.fail()
|
|
|
|
if not isolate:
|
|
cls(constraint, isolate_from_table=False)
|
|
self.assert_compile(
|
|
schema.CreateTable(t),
|
|
"CREATE TABLE tbl (a INTEGER, b INTEGER, "
|
|
"CONSTRAINT my_test_constraint CHECK (a < b) "
|
|
"DEFERRABLE INITIALLY DEFERRED)",
|
|
)
|
|
else:
|
|
cls(constraint)
|
|
self.assert_compile(
|
|
schema.CreateTable(t),
|
|
"CREATE TABLE tbl (a INTEGER, b INTEGER)",
|
|
)
|
|
|
|
def test_render_drop_constraint(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = CheckConstraint(
|
|
"a < b",
|
|
name="my_test_constraint",
|
|
deferrable=True,
|
|
initially="DEFERRED",
|
|
table=t,
|
|
)
|
|
|
|
self.assert_compile(
|
|
schema.DropConstraint(constraint),
|
|
"ALTER TABLE tbl DROP CONSTRAINT my_test_constraint",
|
|
)
|
|
|
|
def test_render_drop_constraint_cascade(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = CheckConstraint(
|
|
"a < b",
|
|
name="my_test_constraint",
|
|
deferrable=True,
|
|
initially="DEFERRED",
|
|
table=t,
|
|
)
|
|
|
|
self.assert_compile(
|
|
schema.DropConstraint(constraint, cascade=True),
|
|
"ALTER TABLE tbl DROP CONSTRAINT my_test_constraint CASCADE",
|
|
)
|
|
|
|
def test_render_drop_constraint_if_exists(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = CheckConstraint("a = 1", name="a1", table=t)
|
|
|
|
self.assert_compile(
|
|
schema.DropConstraint(constraint, if_exists=True),
|
|
"ALTER TABLE tbl DROP CONSTRAINT IF EXISTS a1",
|
|
)
|
|
|
|
def test_render_add_fk_constraint_stringcol(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = ForeignKeyConstraint(["b"], ["t2.a"])
|
|
t.append_constraint(constraint)
|
|
self.assert_compile(
|
|
schema.AddConstraint(constraint),
|
|
"ALTER TABLE tbl ADD FOREIGN KEY(b) REFERENCES t2 (a)",
|
|
)
|
|
|
|
def test_render_add_fk_constraint_realcol(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = ForeignKeyConstraint([t.c.a], [t2.c.b])
|
|
t.append_constraint(constraint)
|
|
self.assert_compile(
|
|
schema.AddConstraint(constraint),
|
|
"ALTER TABLE tbl ADD FOREIGN KEY(a) REFERENCES t2 (b)",
|
|
)
|
|
|
|
def test_render_add_uq_constraint_stringcol(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = UniqueConstraint("a", "b", name="uq_cst")
|
|
t2.append_constraint(constraint)
|
|
self.assert_compile(
|
|
schema.AddConstraint(constraint),
|
|
"ALTER TABLE t2 ADD CONSTRAINT uq_cst UNIQUE (a, b)",
|
|
)
|
|
|
|
def test_render_add_uq_constraint_realcol(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = UniqueConstraint(t2.c.a, t2.c.b, name="uq_cs2")
|
|
self.assert_compile(
|
|
schema.AddConstraint(constraint),
|
|
"ALTER TABLE t2 ADD CONSTRAINT uq_cs2 UNIQUE (a, b)",
|
|
)
|
|
|
|
def test_render_add_pk_constraint(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
assert t.c.a.primary_key is False
|
|
constraint = PrimaryKeyConstraint(t.c.a)
|
|
assert t.c.a.primary_key is True
|
|
self.assert_compile(
|
|
schema.AddConstraint(constraint),
|
|
"ALTER TABLE tbl ADD PRIMARY KEY (a)",
|
|
)
|
|
|
|
def test_render_check_constraint_sql_literal(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = CheckConstraint(t.c.a > 5)
|
|
|
|
self.assert_compile(
|
|
schema.AddConstraint(constraint),
|
|
"ALTER TABLE tbl ADD CHECK (a > 5)",
|
|
)
|
|
|
|
def test_render_check_constraint_inline_sql_literal(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
m = MetaData()
|
|
t = Table(
|
|
"t",
|
|
m,
|
|
Column("a", Integer, CheckConstraint(Column("a", Integer) > 5)),
|
|
)
|
|
|
|
self.assert_compile(
|
|
schema.CreateColumn(t.c.a), "a INTEGER CHECK (a > 5)"
|
|
)
|
|
|
|
def test_render_index_sql_literal(self):
|
|
t, t2 = self._constraint_create_fixture()
|
|
|
|
constraint = Index("name", t.c.a + 5)
|
|
|
|
self.assert_compile(
|
|
schema.CreateIndex(constraint), "CREATE INDEX name ON tbl (a + 5)"
|
|
)
|