@delete_from_all_tables.for_db("mysql", "mariadb")
-def _delete_from_all_tables(cfg, connection, metadata):
+def _delete_from_all_tables(connection, cfg, metadata):
connection.exec_driver_sql("SET foreign_key_checks = 0")
try:
- delete_from_all_tables.call_original(cfg, connection, metadata)
+ delete_from_all_tables.call_original(connection, cfg, metadata)
finally:
connection.exec_driver_sql("SET foreign_key_checks = 1")
):
with self.bind.begin() as conn:
provision.delete_from_all_tables(
- config, conn, self._tables_metadata
+ conn, config, self._tables_metadata
)
@classmethod
from . import util
from .. import exc
from .. import inspect
+from ..engine import Connection
+from ..engine import Engine
from ..engine import url as sa_url
from ..schema import sort_tables_and_constraints
from ..sql import ddl
url = sa_url.make_url(cfg)
elif isinstance(cfg, sa_url.URL):
url = cfg
+ elif isinstance(cfg, (Engine, Connection)):
+ url = cfg.engine.url
else:
url = cfg.db.url
backend = url.get_backend_name()
@register.init
-def delete_from_all_tables(cfg, connection, metadata):
+def delete_from_all_tables(connection, cfg, metadata):
"""an absolutely foolproof delete from all tables routine.
dialects should override this to add special instructions like
*list(_combinations()), argnames="name,parent,child,direction", id_="saaa"
)
class ABCTest(fixtures.MappedTest):
+ __requires__ = ("foreign_key_cycles_w_cascade",)
+
@classmethod
def define_tables(cls, metadata):
parent, child, direction = cls.parent, cls.child, cls.direction
Column(
"child_id",
Integer,
- ForeignKey("%s.id" % child, use_alter=True, name="foo"),
+ ForeignKey(
+ "%s.id" % child,
+ use_alter=True,
+ name="foo",
+ ondelete="cascade",
+ ),
)
)
elif "a" == child and direction == ONETOMANY:
Column(
"parent_id",
Integer,
- ForeignKey("%s.id" % parent, use_alter=True, name="foo"),
+ ForeignKey(
+ "%s.id" % parent,
+ use_alter=True,
+ name="foo",
+ ondelete="cascade",
+ ),
)
)
ta = Table(*ta)
Column(
"child_id",
Integer,
- ForeignKey("%s.id" % child, use_alter=True, name="foo"),
+ ForeignKey(
+ "%s.id" % child,
+ use_alter=True,
+ name="foo",
+ ondelete="cascade",
+ ),
)
)
elif "b" == child and direction == ONETOMANY:
Column(
"parent_id",
Integer,
- ForeignKey("%s.id" % parent, use_alter=True, name="foo"),
+ ForeignKey(
+ "%s.id" % parent,
+ use_alter=True,
+ name="foo",
+ ondelete="cascade",
+ ),
)
)
tb = Table(*tb)
Column(
"child_id",
Integer,
- ForeignKey("%s.id" % child, use_alter=True, name="foo"),
+ ForeignKey(
+ "%s.id" % child,
+ use_alter=True,
+ name="foo",
+ ondelete="cascade",
+ ),
)
)
elif "c" == child and direction == ONETOMANY:
Column(
"parent_id",
Integer,
- ForeignKey("%s.id" % parent, use_alter=True, name="foo"),
+ ForeignKey(
+ "%s.id" % parent,
+ use_alter=True,
+ name="foo",
+ ondelete="cascade",
+ ),
)
)
tc = Table(*tc)
onupdate="im the update",
),
mysql_engine="MyISAM",
+ mariadb_engine="MyISAM",
)
st = Table(
),
Column("data", String(50)),
mysql_engine="MyISAM",
+ mariadb_engine="MyISAM",
)
if testing.against("postgresql", "oracle"):
return skip_if(no_support("sqlite", "not supported by database"))
+ @property
+ def foreign_key_cycles_w_cascade(self):
+ return skip_if(no_support("mssql", "not supported"))
+
@property
def foreign_keys_reflect_as_index(self):
return only_on(["mysql", "mariadb"])