from .schema import ForeignKeyConstraint # noqa
from .schema import Index # noqa
from .schema import MetaData # noqa
-from .schema import PassiveDefault # noqa
from .schema import PrimaryKeyConstraint # noqa
from .schema import Sequence # noqa
from .schema import Table # noqa
from .types import VARBINARY # noqa
from .types import VARCHAR # noqa
-# these are placed last because there are
-# cross dependencies between sqlalchemy.sql and
-# sqlalchemy.engine that cause import cycles
from .engine import create_engine # noqa nosort
from .engine import engine_from_config # noqa nosort
from .engine import create_mock_engine # noqa nosort
+# these are placed last because there are
+# cross dependencies between sqlalchemy.sql and
+# sqlalchemy.engine that cause import cycles
+
__version__ = "1.4.0b1"
from .sql.schema import ForeignKeyConstraint # noqa
from .sql.schema import Index # noqa
from .sql.schema import MetaData # noqa
-from .sql.schema import PassiveDefault # noqa
from .sql.schema import PrimaryKeyConstraint # noqa
from .sql.schema import SchemaItem # noqa
from .sql.schema import Sequence # noqa
from .base import Executable
from .base import SchemaVisitor
from .elements import ClauseElement
-from .. import event
from .. import exc
from .. import util
from ..util import topological
else:
bind.engine.logger.info("DDL execution skipped, criteria not met.")
- @util.deprecated(
- "0.7",
- "The :meth:`.DDLElement.execute_at` method is deprecated and will "
- "be removed in a future release. Please use the :class:`.DDLEvents` "
- "listener interface in conjunction with the "
- ":meth:`.DDLElement.execute_if` method.",
- )
- def execute_at(self, event_name, target):
- """Link execution of this DDL to the DDL lifecycle of a SchemaItem.
-
- Links this ``DDLElement`` to a ``Table`` or ``MetaData`` instance,
- executing it when that schema item is created or dropped. The DDL
- statement will be executed using the same Connection and transactional
- context as the Table create/drop itself. The ``.bind`` property of
- this statement is ignored.
-
- :param event:
- One of the events defined in the schema item's ``.ddl_events``;
- e.g. 'before-create', 'after-create', 'before-drop' or 'after-drop'
-
- :param target:
- The Table or MetaData instance for which this DDLElement will
- be associated with.
-
- A DDLElement instance can be linked to any number of schema items.
-
- ``execute_at`` builds on the ``append_ddl_listener`` interface of
- :class:`.MetaData` and :class:`.Table` objects.
-
- Caveat: Creating or dropping a Table in isolation will also trigger
- any DDL set to ``execute_at`` that Table's MetaData. This may change
- in a future release.
-
- """
-
- def call_event(target, connection, **kw):
- if self._should_execute_deprecated(
- event_name, target, connection, **kw
- ):
- return connection.execute(self.against(target))
-
- event.listen(target, "" + event_name.replace("-", "_"), call_event)
-
@_generative
def against(self, target):
"""Return a copy of this DDL against a specific schema item."""
self.state = state
def _should_execute(self, target, bind, **kw):
- if self.on is not None and not self._should_execute_deprecated(
- None, target, bind, **kw
- ):
- return False
-
if isinstance(self.dialect, util.string_types):
if self.dialect != bind.engine.name:
return False
return True
- def _should_execute_deprecated(self, event, target, bind, **kw):
- if self.on is None:
- return True
- elif isinstance(self.on, util.string_types):
- return self.on == bind.engine.name
- elif isinstance(self.on, (tuple, list, set)):
- return bind.engine.name in self.on
- else:
- return self.on(self, event, target, bind, **kw)
-
def __call__(self, target, bind, **kw):
"""Execute the DDL as a ddl_listener."""
if self._should_execute(target, bind, **kw):
return bind.execute(self.against(target))
- def _check_ddl_on(self, on):
- if on is not None and (
- not isinstance(on, util.string_types + (tuple, list, set))
- and not callable(on)
- ):
- raise exc.ArgumentError(
- "Expected the name of a database dialect, a tuple "
- "of names, or a callable for "
- "'on' criteria, got type '%s'." % type(on).__name__
- )
-
def bind(self):
if self._bind:
return self._bind
__visit_name__ = "ddl"
- @util.deprecated_params(
- on=(
- "0.7",
- "The :paramref:`.DDL.on` parameter is deprecated and will be "
- "removed in a future release. Please refer to "
- ":meth:`.DDLElement.execute_if`.",
- )
- )
- def __init__(self, statement, on=None, context=None, bind=None):
+ def __init__(self, statement, context=None, bind=None):
"""Create a DDL statement.
:param statement:
SQL bind parameters are not available in DDL statements.
- :param on:
-
- Optional filtering criteria. May be a string, tuple or a callable
- predicate. If a string, it will be compared to the name of the
- executing database dialect::
-
- DDL('something', on='postgresql')
-
- If a tuple, specifies multiple dialect names::
-
- DDL('something', on=('postgresql', 'mysql'))
-
- If a callable, it will be invoked with four positional arguments
- as well as optional keyword arguments:
-
- :ddl:
- This DDL element.
-
- :event:
- The name of the event that has triggered this DDL, such as
- 'after-create' Will be None if the DDL is executed explicitly.
-
- :target:
- The ``Table`` or ``MetaData`` object which is the target of
- this event. May be None if the DDL is executed explicitly.
-
- :connection:
- The ``Connection`` being used for DDL execution
-
- :tables:
- Optional keyword argument - a list of Table objects which are to
- be created/ dropped within a MetaData.create_all() or drop_all()
- method call.
-
-
- If the callable returns a true value, the DDL statement will be
- executed.
-
:param context:
Optional dictionary, defaults to None. These values will be
available for use in string substitutions on the DDL statement.
self.statement = statement
self.context = context or {}
- self._check_ddl_on(on)
- self.on = on
self._bind = bind
def __repr__(self):
"""
- def __init__(self, element, on=None, bind=None):
+ def __init__(self, element, bind=None):
self.element = element
- self._check_ddl_on(on)
- self.on = on
self.bind = bind
def _create_rule_disable(self, compiler):
__visit_name__ = "create_table"
def __init__(
- self, element, on=None, bind=None, include_foreign_key_constraints=None
+ self, element, bind=None, include_foreign_key_constraints=None
):
"""Create a :class:`.CreateTable` construct.
.. versionadded:: 1.0.0
"""
- super(CreateTable, self).__init__(element, on=on, bind=bind)
+ super(CreateTable, self).__init__(element, bind=bind)
self.columns = [CreateColumn(column) for column in element.columns]
self.include_foreign_key_constraints = include_foreign_key_constraints
def __repr__(self):
return util.generic_repr(self, omit_kwarg=["info"])
- @property
- @util.deprecated(
- "0.9",
- "The :attr:`.SchemaItem.quote` attribute is deprecated and will be "
- "removed in a future release. Use the :attr:`.quoted_name.quote` "
- "attribute on the ``name`` field of the target schema item to retrieve"
- "quoted status.",
- )
- def quote(self):
- """Return the value of the ``quote`` flag passed
- to this schema object, for those schema items which
- have a ``name`` field.
-
- """
-
- return self.name.quote
-
@util.memoized_property
def info(self):
"""Info dictionary associated with the object, allowing user-defined
__visit_name__ = "table"
- @util.deprecated_params(
- useexisting=(
- "0.7",
- "The :paramref:`.Table.useexisting` parameter is deprecated and "
- "will be removed in a future release. Please use "
- ":paramref:`.Table.extend_existing`.",
- )
- )
def __new__(cls, *args, **kw):
if not args:
# python3k pickle seems to call this
schema = None
keep_existing = kw.pop("keep_existing", False)
extend_existing = kw.pop("extend_existing", False)
- if "useexisting" in kw:
- if extend_existing:
- msg = "useexisting is synonymous with extend_existing."
- raise exc.ArgumentError(msg)
- extend_existing = kw.pop("useexisting", False)
if keep_existing and extend_existing:
msg = "keep_existing and extend_existing are mutually exclusive."
with util.safe_reraise():
metadata._remove_table(name, schema)
- @property
- @util.deprecated(
- "0.9",
- "The :meth:`.SchemaItem.quote` method is deprecated and will be "
- "removed in a future release. Use the :attr:`.quoted_name.quote` "
- "attribute on the ``schema`` field of the target schema item to "
- "retrieve quoted status.",
- )
- def quote_schema(self):
- """Return the value of the ``quote_schema`` flag passed
- to this :class:`.Table`.
- """
-
- return self.schema.quote
-
def __init__(self, *args, **kw):
"""Constructor for :class:`~.schema.Table`.
constraint._set_parent_with_dispatch(self)
- @util.deprecated(
- "0.7",
- "the :meth:`.Table.append_ddl_listener` method is deprecated and "
- "will be removed in a future release. Please refer to "
- ":class:`.DDLEvents`.",
- )
- def append_ddl_listener(self, event_name, listener):
- """Append a DDL event listener to this ``Table``.
-
- """
-
- def adapt_listener(target, connection, **kw):
- listener(event_name, target, connection)
-
- event.listen(self, "" + event_name.replace("-", "_"), adapt_listener)
-
def _set_parent(self, metadata):
metadata._add_table(self.name, self.schema, self)
self.metadata = metadata
return "DefaultClause(%r, for_update=%r)" % (self.arg, self.for_update)
-@util.deprecated_cls(
- "0.6",
- ":class:`.PassiveDefault` is deprecated and will be removed in a "
- "future release. Please refer to :class:`.DefaultClause`.",
-)
-class PassiveDefault(DefaultClause):
- """A DDL-specified DEFAULT column value.
- """
-
- def __init__(self, *arg, **kw):
- DefaultClause.__init__(self, *arg, **kw)
-
-
class Constraint(DialectKWArgs, SchemaItem):
"""A table-level SQL constraint."""
__visit_name__ = "metadata"
- @util.deprecated_params(
- reflect=(
- "0.8",
- "The :paramref:`.MetaData.reflect` flag is deprecated and will "
- "be removed in a future release. Please use the "
- ":meth:`.MetaData.reflect` method.",
- )
- )
def __init__(
self,
bind=None,
- reflect=False,
schema=None,
quote_schema=None,
naming_convention=None,
self._fk_memos = collections.defaultdict(list)
self.bind = bind
- if reflect:
- if not bind:
- raise exc.ArgumentError(
- "A bind must be supplied in conjunction "
- "with reflect=True"
- )
- self.reflect()
tables = None
"""A dictionary of :class:`.Table` objects keyed to their name or "table key".
except exc.UnreflectableTableError as uerr:
util.warn("Skipping table %s: %s" % (name, uerr))
- @util.deprecated(
- "0.7",
- "the :meth:`.MetaData.append_ddl_listener` method is deprecated and "
- "will be removed in a future release. Please refer to "
- ":class:`.DDLEvents`.",
- )
- def append_ddl_listener(self, event_name, listener):
- """Append a DDL event listener to this ``MetaData``.
-
-
- """
-
- def adapt_listener(target, connection, **kw):
- tables = kw["tables"]
- listener(event, target, connection, tables=tables)
-
- event.listen(self, "" + event_name.replace("-", "_"), adapt_listener)
-
def create_all(self, bind=None, tables=None, checkfirst=True):
"""Create all tables stored in this metadata.
assert "xyzzy" in strings
assert "fnord" in strings
- @testing.uses_deprecated(r".*use the DDLEvents")
- def test_table_by_metadata_deprecated(self):
- metadata, users, engine = self.metadata, self.users, self.engine
- DDL("mxyzptlk").execute_at("before-create", users)
- DDL("klptzyxm").execute_at("after-create", users)
- DDL("xyzzy").execute_at("before-drop", users)
- DDL("fnord").execute_at("after-drop", users)
-
- metadata.create_all()
- strings = [str(x) for x in engine.mock]
- assert "mxyzptlk" in strings
- assert "klptzyxm" in strings
- assert "xyzzy" not in strings
- assert "fnord" not in strings
- del engine.mock[:]
- metadata.drop_all()
- strings = [str(x) for x in engine.mock]
- assert "mxyzptlk" not in strings
- assert "klptzyxm" not in strings
- assert "xyzzy" in strings
- assert "fnord" in strings
-
def test_metadata(self):
metadata, engine = self.metadata, self.engine
assert "xyzzy" in strings
assert "fnord" in strings
- @testing.uses_deprecated(r".*use the DDLEvents")
- def test_metadata_deprecated(self):
- metadata, engine = self.metadata, self.engine
-
- DDL("mxyzptlk").execute_at("before-create", metadata)
- DDL("klptzyxm").execute_at("after-create", metadata)
- DDL("xyzzy").execute_at("before-drop", metadata)
- DDL("fnord").execute_at("after-drop", metadata)
-
- metadata.create_all()
- strings = [str(x) for x in engine.mock]
- assert "mxyzptlk" in strings
- assert "klptzyxm" in strings
- assert "xyzzy" not in strings
- assert "fnord" not in strings
- del engine.mock[:]
- metadata.drop_all()
- strings = [str(x) for x in engine.mock]
- assert "mxyzptlk" not in strings
- assert "klptzyxm" not in strings
- assert "xyzzy" in strings
- assert "fnord" in strings
-
def test_conditional_constraint(self):
metadata, users = self.metadata, self.users
nonpg_mock = engines.mock_engine(dialect_name="sqlite")
strings = " ".join(str(x) for x in pg_mock.mock)
assert "my_test_constraint" in strings
- @testing.uses_deprecated(r".*use the DDLEvents")
- def test_conditional_constraint_deprecated(self):
- metadata, users = self.metadata, self.users
- nonpg_mock = engines.mock_engine(dialect_name="sqlite")
- pg_mock = engines.mock_engine(dialect_name="postgresql")
- constraint = CheckConstraint(
- "a < b", name="my_test_constraint", table=users
- )
-
- # by placing the constraint in an Add/Drop construct, the
- # 'inline_ddl' flag is set to False
-
- AddConstraint(constraint, on="postgresql").execute_at(
- "after-create", users
- )
- DropConstraint(constraint, on="postgresql").execute_at(
- "before-drop", users
- )
- metadata.create_all(bind=nonpg_mock)
- strings = " ".join(str(x) for x in nonpg_mock.mock)
- assert "my_test_constraint" not in strings
- metadata.drop_all(bind=nonpg_mock)
- strings = " ".join(str(x) for x in nonpg_mock.mock)
- assert "my_test_constraint" not in strings
- metadata.create_all(bind=pg_mock)
- strings = " ".join(str(x) for x in pg_mock.mock)
- assert "my_test_constraint" in strings
- metadata.drop_all(bind=pg_mock)
- strings = " ".join(str(x) for x in pg_mock.mock)
- assert "my_test_constraint" in strings
-
@testing.requires.sqlite
def test_ddl_execute(self):
engine = create_engine("sqlite:///")
def executor(*a, **kw):
return None
- engine = create_mock_engine(
- testing.db.name + "://", executor
- )
+ engine = create_mock_engine(testing.db.name + "://", executor)
# fmt: off
engine.dialect.identifier_preparer = \
tsa.sql.compiler.IdentifierPreparer(
e = engines.testing_engine(options={"poolclass": AssertionPool})
fn(e)
- @testing.uses_deprecated()
- def test_reflect_uses_bind_constructor_conn(self):
- self._test_reflect_uses_bind(
- lambda e: MetaData(e.connect(), reflect=True)
- )
-
- @testing.uses_deprecated()
- def test_reflect_uses_bind_constructor_engine(self):
- self._test_reflect_uses_bind(lambda e: MetaData(e, reflect=True))
-
def test_reflect_uses_bind_constructor_conn_reflect(self):
self._test_reflect_uses_bind(lambda e: MetaData(e.connect()).reflect())
from sqlalchemy import util
from sqlalchemy import VARCHAR
from sqlalchemy.engine import default
-from sqlalchemy.schema import DDL
from sqlalchemy.sql import coercions
from sqlalchemy.sql import quoted_name
from sqlalchemy.sql import roles
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
-from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import in_
):
select([column("x")], for_update=True)
- @testing.provide_metadata
- def test_table_useexisting(self):
- meta = self.metadata
-
- Table("t", meta, Column("x", Integer))
- meta.create_all()
-
- with testing.expect_deprecated(
- "The Table.useexisting parameter is deprecated and "
- "will be removed in a future release."
- ):
- Table("t", meta, useexisting=True, autoload_with=testing.db)
-
- with testing.expect_deprecated(
- "The Table.useexisting parameter is deprecated and "
- "will be removed in a future release."
- ):
- assert_raises_message(
- exc.ArgumentError,
- "useexisting is synonymous with extend_existing.",
- Table,
- "t",
- meta,
- useexisting=True,
- extend_existing=True,
- autoload_with=testing.db,
- )
-
-
-class DDLListenerDeprecationsTest(fixtures.TestBase):
- def setup(self):
- self.bind = self.engine = engines.mock_engine()
- self.metadata = MetaData(self.bind)
- self.table = Table("t", self.metadata, Column("id", Integer))
- self.users = Table(
- "users",
- self.metadata,
- Column("user_id", Integer, primary_key=True),
- Column("user_name", String(40)),
- )
-
- def test_append_listener(self):
- metadata, table = self.metadata, self.table
-
- def fn(*a):
- return None
-
- with testing.expect_deprecated(".* is deprecated .*"):
- table.append_ddl_listener("before-create", fn)
- with testing.expect_deprecated(".* is deprecated .*"):
- assert_raises(
- exc.InvalidRequestError, table.append_ddl_listener, "blah", fn
- )
-
- with testing.expect_deprecated(".* is deprecated .*"):
- metadata.append_ddl_listener("before-create", fn)
- with testing.expect_deprecated(".* is deprecated .*"):
- assert_raises(
- exc.InvalidRequestError,
- metadata.append_ddl_listener,
- "blah",
- fn,
- )
-
- def test_deprecated_append_ddl_listener_table(self):
- metadata, users, engine = self.metadata, self.users, self.engine
- canary = []
- with testing.expect_deprecated(".* is deprecated .*"):
- users.append_ddl_listener(
- "before-create", lambda e, t, b: canary.append("mxyzptlk")
- )
- with testing.expect_deprecated(".* is deprecated .*"):
- users.append_ddl_listener(
- "after-create", lambda e, t, b: canary.append("klptzyxm")
- )
- with testing.expect_deprecated(".* is deprecated .*"):
- users.append_ddl_listener(
- "before-drop", lambda e, t, b: canary.append("xyzzy")
- )
- with testing.expect_deprecated(".* is deprecated .*"):
- users.append_ddl_listener(
- "after-drop", lambda e, t, b: canary.append("fnord")
- )
-
- metadata.create_all()
- assert "mxyzptlk" in canary
- assert "klptzyxm" in canary
- assert "xyzzy" not in canary
- assert "fnord" not in canary
- del engine.mock[:]
- canary[:] = []
- metadata.drop_all()
- assert "mxyzptlk" not in canary
- assert "klptzyxm" not in canary
- assert "xyzzy" in canary
- assert "fnord" in canary
-
- def test_deprecated_append_ddl_listener_metadata(self):
- metadata, engine = self.metadata, self.engine
- canary = []
- with testing.expect_deprecated(".* is deprecated .*"):
- metadata.append_ddl_listener(
- "before-create",
- lambda e, t, b, tables=None: canary.append("mxyzptlk"),
- )
- with testing.expect_deprecated(".* is deprecated .*"):
- metadata.append_ddl_listener(
- "after-create",
- lambda e, t, b, tables=None: canary.append("klptzyxm"),
- )
- with testing.expect_deprecated(".* is deprecated .*"):
- metadata.append_ddl_listener(
- "before-drop",
- lambda e, t, b, tables=None: canary.append("xyzzy"),
- )
- with testing.expect_deprecated(".* is deprecated .*"):
- metadata.append_ddl_listener(
- "after-drop",
- lambda e, t, b, tables=None: canary.append("fnord"),
- )
-
- metadata.create_all()
- assert "mxyzptlk" in canary
- assert "klptzyxm" in canary
- assert "xyzzy" not in canary
- assert "fnord" not in canary
- del engine.mock[:]
- canary[:] = []
- metadata.drop_all()
- assert "mxyzptlk" not in canary
- assert "klptzyxm" not in canary
- assert "xyzzy" in canary
- assert "fnord" in canary
-
- def test_filter_deprecated(self):
- cx = self.engine
-
- tbl = Table("t", MetaData(), Column("id", Integer))
- target = cx.name
-
- assert DDL("")._should_execute_deprecated("x", tbl, cx)
- with testing.expect_deprecated(".* is deprecated .*"):
- assert DDL("", on=target)._should_execute_deprecated("x", tbl, cx)
- with testing.expect_deprecated(".* is deprecated .*"):
- assert not DDL("", on="bogus")._should_execute_deprecated(
- "x", tbl, cx
- )
- with testing.expect_deprecated(".* is deprecated .*"):
- assert DDL(
- "", on=lambda d, x, y, z: True
- )._should_execute_deprecated("x", tbl, cx)
- with testing.expect_deprecated(".* is deprecated .*"):
- assert DDL(
- "", on=lambda d, x, y, z: z.engine.name != "bogus"
- )._should_execute_deprecated("x", tbl, cx)
-
class ConvertUnicodeDeprecationTest(fixtures.TestBase):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
