as well as related mechanisms to look for bytestrings in
DBAPI ``cursor.description`` etc. have been removed.
+ * The ``.bind`` attribute and parameter from :class:`.MetaData`,
+ :class:`.Table`, and from all DDL/DML/DQL elements that previously could
+ refer to a "bound engine"
+
* More are in progress as development continues
:ref:`deprecation_20_mode` - specific guidelines on how to use
"2.0 deprecations mode" in SQLAlchemy 1.4.
-.. _error_c9bf:
-
-A bind was located via legacy bound metadata, but since future=True is set on this Session, this bind is ignored.
--------------------------------------------------------------------------------------------------------------------
-
-The concept of "bound metadata" is being removed in SQLAlchemy 2.0. This
-refers to the :paramref:`_schema.MetaData.bind` parameter on the
-:class:`_schema.MetaData` object that in turn allows objects like the ORM
-:class:`_orm.Session` to associate a particular mapped class with an
-:class:`_orm.Engine`. In SQLAlchemy 2.0, the :class:`_orm.Session` must be
-linked to each :class:`_orm.Engine` directly. That is, instead of instantiating
-the :class:`_orm.Session` or
-:class:`_orm.sessionmaker` without any arguments, and associating the
-:class:`_engine.Engine` with the :class:`_schema.MetaData`::
-
- engine = create_engine("sqlite://")
- Session = sessionmaker()
- metadata_obj = MetaData(bind=engine)
- Base = declarative_base(metadata=metadata_obj)
-
- class MyClass(Base):
- # ...
-
-
- session = Session()
- session.add(MyClass())
- session.commit()
-
-The :class:`_engine.Engine` must instead be associated directly with the
-:class:`_orm.sessionmaker` or :class:`_orm.Session`. The
-:class:`_schema.MetaData` object should no longer be associated with any
-engine::
-
-
- engine = create_engine("sqlite://")
- Session = sessionmaker(engine)
- Base = declarative_base()
-
- class MyClass(Base):
- # ...
-
-
- session = Session()
- session.add(MyClass())
- session.commit()
-
-In SQLAlchemy 1.4, this :term:`2.0 style` behavior is enabled when the
-:paramref:`_orm.Session.future` flag is set on :class:`_orm.sessionmaker`
-or :class:`_orm.Session`.
-
.. _error_s9r1:
Object is being merged into a Session along the backref cascade
+Legacy Exceptions
+=================
+
+Exceptions in this section are not generated by current SQLAlchemy
+versions, however are provided here to suit exception message hyperlinks.
+
+
+.. _error_c9bf:
+
+A bind was located via legacy bound metadata, but since future=True is set on this Session, this bind is ignored.
+-------------------------------------------------------------------------------------------------------------------
+
+.. note:: This is a legacy error message that is not in the 2.x series of
+ SQLAlchemy.
+
+The concept of "bound metadata" is being removed in SQLAlchemy 2.0. This
+refers to the :paramref:`_schema.MetaData.bind` parameter on the
+:class:`_schema.MetaData` object that in turn allows objects like the ORM
+:class:`_orm.Session` to associate a particular mapped class with an
+:class:`_orm.Engine`. In SQLAlchemy 2.0, the :class:`_orm.Session` must be
+linked to each :class:`_orm.Engine` directly. That is, instead of instantiating
+the :class:`_orm.Session` or
+:class:`_orm.sessionmaker` without any arguments, and associating the
+:class:`_engine.Engine` with the :class:`_schema.MetaData`::
+
+ engine = create_engine("sqlite://")
+ Session = sessionmaker()
+ metadata_obj = MetaData(bind=engine)
+ Base = declarative_base(metadata=metadata_obj)
+
+ class MyClass(Base):
+ # ...
+
+
+ session = Session()
+ session.add(MyClass())
+ session.commit()
+
+The :class:`_engine.Engine` must instead be associated directly with the
+:class:`_orm.sessionmaker` or :class:`_orm.Session`. The
+:class:`_schema.MetaData` object should no longer be associated with any
+engine::
+
+
+ engine = create_engine("sqlite://")
+ Session = sessionmaker(engine)
+ Base = declarative_base()
+
+ class MyClass(Base):
+ # ...
+
+
+ session = Session()
+ session.add(MyClass())
+ session.commit()
+
+In SQLAlchemy 1.4, this :term:`2.0 style` behavior is enabled when the
+:paramref:`_orm.Session.future` flag is set on :class:`_orm.sessionmaker`
+or :class:`_orm.Session`.
+
from .schema import PrimaryKeyConstraint
from .schema import Sequence
from .schema import Table
-from .schema import ThreadLocalMetaData
from .schema import UniqueConstraint
from .sql import alias
from .sql import all_
def declarative_base(
- bind=None,
metadata=None,
mapper=None,
cls=object,
``sqlalchemy.orm`` package from the ``declarative.ext`` package.
- :param bind: An optional
- :class:`~sqlalchemy.engine.Connectable`, will be assigned
- the ``bind`` attribute on the :class:`~sqlalchemy.schema.MetaData`
- instance.
-
- .. deprecated:: 1.4 The "bind" argument to declarative_base is
- deprecated and will be removed in SQLAlchemy 2.0.
-
:param metadata:
An optional :class:`~sqlalchemy.schema.MetaData` instance. All
:class:`~sqlalchemy.schema.Table` objects implicitly declared by
"""
- if bind is not None:
- # util.deprecated_params does not work
- util.warn_deprecated_20(
- "The ``bind`` argument to declarative_base is "
- "deprecated and will be removed in SQLAlchemy 2.0.",
- )
-
return registry(
- _bind=bind,
metadata=metadata,
class_registry=class_registry,
constructor=constructor,
metadata=None,
class_registry=None,
constructor=_declarative_constructor,
- _bind=None,
):
r"""Construct a new :class:`_orm.registry`
"""
lcl_metadata = metadata or MetaData()
- if _bind:
- lcl_metadata.bind = _bind
if class_registry is None:
class_registry = weakref.WeakValueDictionary()
mapperlib._legacy_registry = registry()
-@util.deprecated_params(
- bind=(
- "2.0",
- "The ``bind`` argument to as_declarative is "
- "deprecated and will be removed in SQLAlchemy 2.0.",
- )
-)
def as_declarative(**kw):
"""
Class decorator which will adapt a given class into a
:meth:`_orm.registry.as_declarative_base`
"""
- bind, metadata, class_registry = (
- kw.pop("bind", None),
+ metadata, class_registry = (
kw.pop("metadata", None),
kw.pop("class_registry", None),
)
return registry(
- _bind=bind, metadata=metadata, class_registry=class_registry
+ metadata=metadata, class_registry=class_registry
).as_declarative_base(**kw)
if self.bind:
return self.bind
- # now we are in legacy territory. looking for "bind" on tables
- # that are via bound metadata. this goes away in 2.0.
-
- future_msg = ""
- future_code = ""
-
- if mapper and clause is None:
- clause = mapper.persist_selectable
-
- if clause is not None:
- if clause.bind:
- if self.future:
- future_msg = (
- " A bind was located via legacy bound metadata, but "
- "since future=True is set on this Session, this "
- "bind is ignored."
- )
- else:
- util.warn_deprecated_20(
- "This Session located a target engine via bound "
- "metadata; as this functionality will be removed in "
- "SQLAlchemy 2.0, an Engine object should be passed "
- "to the Session() constructor directly."
- )
- return clause.bind
-
- if mapper:
- if mapper.persist_selectable.bind:
- if self.future:
- future_msg = (
- " A bind was located via legacy bound metadata, but "
- "since future=True is set on this Session, this "
- "bind is ignored."
- )
- else:
- util.warn_deprecated_20(
- "This Session located a target engine via bound "
- "metadata; as this functionality will be removed in "
- "SQLAlchemy 2.0, an Engine object should be passed "
- "to the Session() constructor directly."
- )
- return mapper.persist_selectable.bind
-
context = []
if mapper is not None:
context.append("mapper %s" % mapper)
context.append("SQL expression")
raise sa_exc.UnboundExecutionError(
- "Could not locate a bind configured on %s or this Session.%s"
- % (", ".join(context), future_msg),
- code=future_code,
+ "Could not locate a bind configured on %s or this Session."
+ % (", ".join(context),),
)
def query(self, *entities, **kwargs):
from .sql.schema import SchemaItem # noqa
from .sql.schema import Sequence # noqa
from .sql.schema import Table # noqa
-from .sql.schema import ThreadLocalMetaData # noqa
from .sql.schema import UniqueConstraint # noqa
"""
return self._execution_options
- @property
- @util.deprecated_20(
- ":attr:`.Executable.bind`",
- alternative="Bound metadata is being removed as of SQLAlchemy 2.0.",
- enable_warnings=False,
- )
- def bind(self):
- """Returns the :class:`_engine.Engine` or :class:`_engine.Connection`
- to
- which this :class:`.Executable` is bound, or None if none found.
-
- This is a traversal which checks locally, then
- checks among the "from" clauses of associated objects
- until a bound engine or connection is found.
-
- """
- if self._bind is not None:
- return self._bind
-
- for f in _from_objects(self):
- if f is self:
- continue
- engine = f.bind
- if engine is not None:
- return engine
- else:
- return None
-
class prefix_anon_map(dict):
"""A map that creates new keys for missing key access.
return hash(tuple(x for x in self))
-def _bind_or_error(schemaitem, msg=None):
-
- util.warn_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL "
- "against an engine or connection will be required in SQLAlchemy 2.0."
- )
- bind = schemaitem.bind
- if not bind:
- name = schemaitem.__class__.__name__
- label = getattr(
- schemaitem, "fullname", getattr(schemaitem, "name", None)
- )
- if label:
- item = "%s object %r" % (name, label)
- else:
- item = "%s object" % name
- if msg is None:
- msg = (
- "%s is not bound to an Engine or Connection. "
- "Execution can not proceed without a database to execute "
- "against." % item
- )
- raise exc.UnboundExecutionError(msg)
- return bind
-
-
def _entity_namespace(entity):
"""Return the nearest .entity_namespace for the given entity.
if self._should_execute(target, bind, **kw):
return bind.execute(self.against(target))
- def bind(self):
- if self._bind:
- return self._bind
-
- def _set_bind(self, bind):
- self._bind = bind
-
- bind = property(bind, _set_bind)
-
def _generate(self):
s = self.__class__.__new__(self.__class__)
s.__dict__ = self.__dict__.copy()
__visit_name__ = "ddl"
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_ddl.DDL.bind` argument is deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
- def __init__(self, statement, context=None, bind=None):
+ def __init__(self, statement, context=None):
"""Create a DDL statement.
:param statement:
Optional dictionary, defaults to None. These values will be
available for use in string substitutions on the DDL statement.
- :param bind:
- Optional. A :class:`.Connectable`, used by
- default when ``execute()`` is invoked without a bind argument.
-
-
.. seealso::
:class:`.DDLEvents`
self.statement = statement
self.context = context or {}
- self._bind = bind
-
def __repr__(self):
return "<%s@%s; %s>" % (
type(self).__name__,
"""
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_ddl.DDLElement.bind` argument is "
- "deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
def __init__(
self,
element,
- bind=None,
if_exists=False,
if_not_exists=False,
- _legacy_bind=None,
):
self.element = element
- if bind:
- self.bind = bind
- elif _legacy_bind:
- self.bind = _legacy_bind
self.if_exists = if_exists
self.if_not_exists = if_not_exists
__visit_name__ = "create_table"
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_ddl.CreateTable.bind` argument is deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
def __init__(
self,
element,
- bind=None,
include_foreign_key_constraints=None,
if_not_exists=False,
):
:param element: a :class:`_schema.Table` that's the subject
of the CREATE
:param on: See the description for 'on' in :class:`.DDL`.
- :param bind: See the description for 'bind' in :class:`.DDL`.
:param include_foreign_key_constraints: optional sequence of
:class:`_schema.ForeignKeyConstraint` objects that will be included
inline within the CREATE construct; if omitted, all foreign key
.. versionadded:: 1.4.0b2
"""
- super(CreateTable, self).__init__(
- element, _legacy_bind=bind, if_not_exists=if_not_exists
- )
+ super(CreateTable, self).__init__(element, if_not_exists=if_not_exists)
self.columns = [CreateColumn(column) for column in element.columns]
self.include_foreign_key_constraints = include_foreign_key_constraints
__visit_name__ = "drop_table"
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_ddl.DropTable.bind` argument is "
- "deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
- def __init__(self, element, bind=None, if_exists=False):
+ def __init__(self, element, if_exists=False):
"""Create a :class:`.DropTable` construct.
:param element: a :class:`_schema.Table` that's the subject
of the DROP.
:param on: See the description for 'on' in :class:`.DDL`.
- :param bind: See the description for 'bind' in :class:`.DDL`.
:param if_exists: if True, an IF EXISTS operator will be applied to the
construct.
.. versionadded:: 1.4.0b2
"""
- super(DropTable, self).__init__(
- element, _legacy_bind=bind, if_exists=if_exists
- )
+ super(DropTable, self).__init__(element, if_exists=if_exists)
class CreateSequence(_CreateDropBase):
__visit_name__ = "create_index"
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_ddl.CreateIndex.bind` argument is "
- "deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
- def __init__(self, element, bind=None, if_not_exists=False):
+ def __init__(self, element, if_not_exists=False):
"""Create a :class:`.Createindex` construct.
:param element: a :class:`_schema.Index` that's the subject
of the CREATE.
:param on: See the description for 'on' in :class:`.DDL`.
- :param bind: See the description for 'bind' in :class:`.DDL`.
:param if_not_exists: if True, an IF NOT EXISTS operator will be
applied to the construct.
.. versionadded:: 1.4.0b2
"""
- super(CreateIndex, self).__init__(
- element, _legacy_bind=bind, if_not_exists=if_not_exists
- )
+ super(CreateIndex, self).__init__(element, if_not_exists=if_not_exists)
class DropIndex(_CreateDropBase):
__visit_name__ = "drop_index"
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_ddl.DropIndex.bind` argument is "
- "deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
- def __init__(self, element, bind=None, if_exists=False):
+ def __init__(self, element, if_exists=False):
"""Create a :class:`.DropIndex` construct.
:param element: a :class:`_schema.Index` that's the subject
of the DROP.
:param on: See the description for 'on' in :class:`.DDL`.
- :param bind: See the description for 'bind' in :class:`.DDL`.
:param if_exists: if True, an IF EXISTS operator will be applied to the
construct.
.. versionadded:: 1.4.0b2
"""
- super(DropIndex, self).__init__(
- element, _legacy_bind=bind, if_exists=if_exists
- )
+ super(DropIndex, self).__init__(element, if_exists=if_exists)
class AddConstraint(_CreateDropBase):
"in SQLAlchemy 2.0. Please refer to the "
":meth:`%(classname)s.values` method."
),
- bind=(
- "The :paramref:`%(func)s.bind` parameter will be removed in "
- "SQLAlchemy 2.0. Please use explicit connection execution."
- ),
inline=(
"The :paramref:`%(func)s.inline` parameter will be "
"removed in "
)
self._validate_dialect_kwargs(dialect_kw)
- def bind(self):
- """Return a 'bind' linked to this :class:`.UpdateBase`
- or a :class:`_schema.Table` associated with it.
-
- """
- return self._bind or self.table.bind
-
- def _set_bind(self, bind):
- self._bind = bind
-
- bind = property(bind, _set_bind)
-
@_generative
def returning(self, *cols):
r"""Add a :term:`RETURNING` or equivalent clause to this statement.
[
"values",
"inline",
- "bind",
"prefixes",
"returning",
"return_defaults",
table,
values=None,
inline=False,
- bind=None,
prefixes=None,
returning=None,
return_defaults=False,
"""
super(Insert, self).__init__(table, values, prefixes)
- self._bind = bind
self._inline = inline
if returning:
self._returning = returning
"whereclause",
"values",
"inline",
- "bind",
"prefixes",
"returning",
"return_defaults",
whereclause=None,
values=None,
inline=False,
- bind=None,
prefixes=None,
returning=None,
return_defaults=False,
"""
self._preserve_parameter_order = preserve_parameter_order
super(Update, self).__init__(table, values, prefixes)
- self._bind = bind
if returning:
self._returning = returning
if whereclause is not None:
@ValuesBase._constructor_20_deprecations(
"delete",
"Delete",
- ["whereclause", "values", "bind", "prefixes", "returning"],
+ ["whereclause", "values", "prefixes", "returning"],
)
def __init__(
self,
table,
whereclause=None,
- bind=None,
returning=None,
prefixes=None,
**dialect_kw
:ref:`deletes` - SQL Expression Tutorial
"""
- self._bind = bind
self.table = coercions.expect(
roles.DMLTableRole, table, apply_propagate_attrs=self
)
dictionary of bind parameter names and values
using the ``params`` accessor.
- :param bind: An ``Engine`` or ``Connection`` from which a
- ``Compiled`` will be acquired. This argument takes precedence over
- this :class:`_expression.ClauseElement`'s bound engine, if any.
+ :param bind: An :class:`.Connection` or :class:`.Engine` which
+ can provide a :class:`.Dialect` in order to generate a
+ :class:`.Compiled` object. If the ``bind`` and
+ ``dialect`` parameters are both omitted, a default SQL compiler
+ is used.
:param column_keys: Used for INSERT and UPDATE statements, a list of
column names which should be present in the VALUES clause of the
compiled statement. If ``None``, all columns from the target table
object are rendered.
- :param dialect: A ``Dialect`` instance from which a ``Compiled``
- will be acquired. This argument takes precedence over the `bind`
- argument as well as this :class:`_expression.ClauseElement`
- 's bound engine,
- if any.
+ :param dialect: A :class:`.Dialect` instance which can generate
+ a :class:`.Compiled` object. This argument takes precedence over
+ the ``bind`` argument.
:param compile_kwargs: optional dictionary of additional parameters
that will be passed through to the compiler within all "visit"
if not dialect:
if bind:
dialect = bind.dialect
- elif self.bind:
- dialect = self.bind.dialect
else:
if self.stringify_dialect == "default":
default = util.preloaded.engine_default
_allow_label_resolve = False
- def __init__(self, text, bind=None):
- self._bind = bind
+ def __init__(self, text):
self._bindparams = {}
def repl(m):
@classmethod
@_document_text_coercion("text", ":func:`.text`", ":paramref:`.text.text`")
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_sql.text.bind` argument is deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
- def _create_text(cls, text, bind=None):
+ def _create_text(cls, text):
r"""Construct a new :class:`_expression.TextClause` clause,
representing
a textual SQL string directly.
to specify bind parameters; they will be compiled to their
engine-specific format.
- :param bind:
- an optional connection or engine to be used for this text query.
-
.. seealso::
:ref:`sqlexpression_text` - in the Core tutorial
"""
- return TextClause(text, bind=bind)
+ return TextClause(text)
@_generative
def bindparams(self, *binds, **names_to_values):
"""
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_sql.text.bind` argument is deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
def __init__(self, name, *clauses, **kw):
"""Construct a :class:`.Function`.
self.packagenames = kw.pop("packagenames", None) or ()
self.name = name
- self._bind = self._get_bind(kw)
self.type = sqltypes.to_instance(kw.get("type_", None))
FunctionElement.__init__(self, *clauses, **kw)
- def _get_bind(self, kw):
- if "bind" in kw:
- util.warn_deprecated_20(
- "The Function.bind argument is deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- )
- return kw["bind"]
-
def _bind_param(self, operator, obj, type_=None, **kw):
return BindParameter(
self.name,
]
self._has_args = self._has_args or bool(parsed_args)
self.packagenames = ()
- self._bind = self._get_bind(kwargs)
self.clause_expr = ClauseList(
operator=operators.comma_op, group_contents=True, *parsed_args
).self_group()
assert isinstance(
seq, schema.Sequence
), "next_value() accepts a Sequence object as input."
- self._bind = self._get_bind(kw)
self.sequence = seq
self.type = sqltypes.to_instance(
seq.data_type or getattr(self, "type", None)
"""
import collections
-import sqlalchemy
from . import coercions
from . import ddl
from . import roles
from . import type_api
from . import visitors
-from .base import _bind_or_error
from .base import DedupeColumnCollection
from .base import DialectKWArgs
from .base import Executable
table. The metadata is used as a point of association of this table
with other tables which are referenced via foreign key. It also
may be used to associate this table with a particular
- :class:`.Connectable`.
+ :class:`.Connection` or :class:`.Engine`.
:param \*args: Additional positional arguments are used primarily
to add the list of :class:`_schema.Column`
resolve_fks=True,
_extend_on=None,
):
- if autoload_with is None:
- autoload_with = _bind_or_error(
- metadata,
- msg="No engine is bound to this Table's MetaData. "
- "Pass an engine to the Table via "
- "autoload_with=<someengine_or_connection>",
- )
-
insp = inspection.inspect(autoload_with)
with insp._inspection_context() as conn_insp:
conn_insp.reflect_table(
def __str__(self):
return _get_table_key(self.description, self.schema)
- @property
- def bind(self):
- """Return the connectable associated with this Table."""
-
- return self.metadata and self.metadata.bind or None
-
def add_is_dependent_on(self, table):
"""Add a 'dependency' for this Table.
metadata._add_table(self.name, self.schema, self)
self.metadata = metadata
- @util.deprecated(
- "1.4",
- "The :meth:`_schema.Table.exists` method is deprecated and will be "
- "removed in a future release. Please refer to "
- ":meth:`_reflection.Inspector.has_table`.",
- )
- def exists(self, bind=None):
- """Return True if this table exists."""
-
- if bind is None:
- bind = _bind_or_error(self)
-
- insp = inspection.inspect(bind)
- return insp.has_table(self.name, schema=self.schema)
-
- def create(self, bind=None, checkfirst=False):
+ def create(self, bind, checkfirst=False):
"""Issue a ``CREATE`` statement for this
- :class:`_schema.Table`, using the given :class:`.Connectable`
+ :class:`_schema.Table`, using the given
+ :class:`.Connection` or :class:`.Engine`
for connectivity.
- .. note:: the "bind" argument will be required in
- SQLAlchemy 2.0.
-
.. seealso::
:meth:`_schema.MetaData.create_all`.
"""
- if bind is None:
- bind = _bind_or_error(self)
bind._run_ddl_visitor(ddl.SchemaGenerator, self, checkfirst=checkfirst)
- def drop(self, bind=None, checkfirst=False):
+ def drop(self, bind, checkfirst=False):
"""Issue a ``DROP`` statement for this
- :class:`_schema.Table`, using the given :class:`.Connectable`
- for connectivity.
-
- .. note:: the "bind" argument will be required in
- SQLAlchemy 2.0.
+ :class:`_schema.Table`, using the given
+ :class:`.Connection` or :class:`.Engine` for connectivity.
.. seealso::
:meth:`_schema.MetaData.drop_all`.
"""
- if bind is None:
- bind = _bind_or_error(self)
bind._run_ddl_visitor(ddl.SchemaDropper, self, checkfirst=checkfirst)
@util.deprecated(
self, distilled_params, execution_options
)
- @property
- def bind(self):
- """Return the connectable associated with this default."""
- if getattr(self, "column", None) is not None:
- return self.column.table.bind
- else:
- return None
-
class ColumnDefault(DefaultGenerator):
"""A plain default value on a column.
for this :class:`.Sequence` within any SQL expression.
"""
- if self.bind:
- return util.preloaded.sql_functions.func.next_value(
- self, bind=self.bind
- )
- else:
- return util.preloaded.sql_functions.func.next_value(self)
+ return util.preloaded.sql_functions.func.next_value(self)
def _set_parent(self, column, **kw):
super(Sequence, self)._set_parent(column)
self.metadata = metadata
self.metadata._sequences[self._key] = self
- @property
- def bind(self):
- if self.metadata:
- return self.metadata.bind
- else:
- return None
-
- def create(self, bind=None, checkfirst=True):
- """Creates this sequence in the database.
-
- .. note:: the "bind" argument will be required in
- SQLAlchemy 2.0.
-
- """
+ def create(self, bind, checkfirst=True):
+ """Creates this sequence in the database."""
- if bind is None:
- bind = _bind_or_error(self)
bind._run_ddl_visitor(ddl.SchemaGenerator, self, checkfirst=checkfirst)
- def drop(self, bind=None, checkfirst=True):
- """Drops this sequence from the database.
+ def drop(self, bind, checkfirst=True):
+ """Drops this sequence from the database."""
- .. note:: the "bind" argument will be required in
- SQLAlchemy 2.0.
-
- """
-
- if bind is None:
- bind = _bind_or_error(self)
bind._run_ddl_visitor(ddl.SchemaDropper, self, checkfirst=checkfirst)
def _not_a_column_expr(self):
for expr, colexpr in zip(expressions, col_expressions)
]
- @property
- def bind(self):
- """Return the connectable associated with this Index."""
-
- return self.table.bind
-
- def create(self, bind=None, checkfirst=False):
+ def create(self, bind, checkfirst=False):
"""Issue a ``CREATE`` statement for this
- :class:`.Index`, using the given :class:`.Connectable`
- for connectivity.
-
- .. note:: the "bind" argument will be required in
- SQLAlchemy 2.0.
+ :class:`.Index`, using the given
+ :class:`.Connection` or :class:`.Engine`` for connectivity.
.. seealso::
:meth:`_schema.MetaData.create_all`.
"""
- if bind is None:
- bind = _bind_or_error(self)
bind._run_ddl_visitor(ddl.SchemaGenerator, self, checkfirst=checkfirst)
return self
- def drop(self, bind=None, checkfirst=False):
+ def drop(self, bind, checkfirst=False):
"""Issue a ``DROP`` statement for this
- :class:`.Index`, using the given :class:`.Connectable`
- for connectivity.
-
- .. note:: the "bind" argument will be required in
- SQLAlchemy 2.0.
+ :class:`.Index`, using the given
+ :class:`.Connection` or :class:`.Engine` for connectivity.
.. seealso::
:meth:`_schema.MetaData.drop_all`.
"""
- if bind is None:
- bind = _bind_or_error(self)
bind._run_ddl_visitor(ddl.SchemaDropper, self, checkfirst=checkfirst)
def __repr__(self):
__visit_name__ = "metadata"
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_schema.MetaData.bind` argument is deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
def __init__(
self,
- bind=None,
schema=None,
quote_schema=None,
naming_convention=None,
):
"""Create a new MetaData object.
- :param bind:
- An Engine or Connection to bind to. May also be a string or URL
- instance, these are passed to :func:`_sa.create_engine` and
- this :class:`_schema.MetaData` will
- be bound to the resulting engine.
-
:param schema:
The default schema to use for the :class:`_schema.Table`,
:class:`.Sequence`, and potentially other objects associated with
self._sequences = {}
self._fk_memos = collections.defaultdict(list)
- self.bind = bind
-
tables = None
"""A dictionary of :class:`_schema.Table`
objects keyed to their name or "table key".
"""
def __repr__(self):
- if self.bind:
- return "MetaData(bind=%r)" % self.bind
- else:
- return "MetaData()"
+ return "MetaData()"
def __contains__(self, table_or_key):
if not isinstance(table_or_key, str):
self.tables = state["tables"]
self.schema = state["schema"]
self.naming_convention = state["naming_convention"]
- self._bind = None
self._sequences = state["sequences"]
self._schemas = state["schemas"]
self._fk_memos = state["fk_memos"]
- def is_bound(self):
- """True if this MetaData is bound to an Engine or Connection."""
-
- return self._bind is not None
-
- def bind(self):
- """An :class:`_engine.Engine` or :class:`_engine.Connection`
- to which this
- :class:`_schema.MetaData` is bound.
-
- Typically, a :class:`_engine.Engine` is assigned to this attribute
- so that "implicit execution" may be used, or alternatively
- as a means of providing engine binding information to an
- ORM :class:`.Session` object::
-
- engine = create_engine("someurl://")
- metadata.bind = engine
-
- .. deprecated :: 1.4
-
- The metadata.bind attribute, as part of the deprecated system
- of "implicit execution", is itself deprecated and will be
- removed in SQLAlchemy 2.0.
-
- .. seealso::
-
- :ref:`dbengine_implicit` - background on "bound metadata"
-
- """
- return self._bind
-
- @util.preload_module("sqlalchemy.engine.url")
- def _bind_to(self, bind):
- """Bind this MetaData to an Engine, Connection, string or URL."""
- url = util.preloaded.engine_url
- if isinstance(bind, (str, url.URL)):
- self._bind = sqlalchemy.create_engine(bind)
- else:
- self._bind = bind
-
- bind = property(bind, _bind_to)
-
def clear(self):
"""Clear all Table objects from this MetaData."""
def reflect(
self,
- bind=None,
+ bind,
schema=None,
views=False,
only=None,
in this ``MetaData`` no longer exists in the database.
:param bind:
- A :class:`.Connectable` used to access the database; if None, uses
- the existing bind on this ``MetaData``, if any.
-
- .. note:: the "bind" argument will be required in
- SQLAlchemy 2.0.
+ A :class:`.Connection` or :class:`.Engine` used to access the
+ database.
:param schema:
Optional, query and reflect tables from an alternate schema.
objects reflected.
"""
- if bind is None:
- bind = _bind_or_error(self)
with inspection.inspect(bind)._inspection_context() as insp:
reflect_opts = {
except exc.UnreflectableTableError as uerr:
util.warn("Skipping table %s: %s" % (name, uerr))
- def create_all(self, bind=None, tables=None, checkfirst=True):
+ def create_all(self, bind, tables=None, checkfirst=True):
"""Create all tables stored in this metadata.
Conditional by default, will not attempt to recreate tables already
present in the target database.
:param bind:
- A :class:`.Connectable` used to access the
- database; if None, uses the existing bind on this ``MetaData``, if
- any.
-
- .. note:: the "bind" argument will be required in
- SQLAlchemy 2.0.
+ A :class:`.Connection` or :class:`.Engine` used to access the
+ database.
:param tables:
Optional list of ``Table`` objects, which is a subset of the total
in the target database.
"""
- if bind is None:
- bind = _bind_or_error(self)
bind._run_ddl_visitor(
ddl.SchemaGenerator, self, checkfirst=checkfirst, tables=tables
)
- def drop_all(self, bind=None, tables=None, checkfirst=True):
+ def drop_all(self, bind, tables=None, checkfirst=True):
"""Drop all tables stored in this metadata.
Conditional by default, will not attempt to drop tables not present in
the target database.
:param bind:
- A :class:`.Connectable` used to access the
- database; if None, uses the existing bind on this ``MetaData``, if
- any.
-
- .. note:: the "bind" argument will be required in
- SQLAlchemy 2.0.
+ A :class:`.Connection` or :class:`.Engine` used to access the
+ database.
:param tables:
Optional list of ``Table`` objects, which is a subset of the
present in the target database.
"""
- if bind is None:
- bind = _bind_or_error(self)
bind._run_ddl_visitor(
ddl.SchemaDropper, self, checkfirst=checkfirst, tables=tables
)
-@util.deprecated_cls(
- "1.4",
- ":class:`.ThreadLocalMetaData` is deprecated and will be removed "
- "in a future release.",
- constructor="__init__",
-)
-class ThreadLocalMetaData(MetaData):
- """A MetaData variant that presents a different ``bind`` in every thread.
-
- Makes the ``bind`` property of the MetaData a thread-local value, allowing
- this collection of tables to be bound to different ``Engine``
- implementations or connections in each thread.
-
- The ThreadLocalMetaData starts off bound to None in each thread. Binds
- must be made explicitly by assigning to the ``bind`` property or using
- ``connect()``. You can also re-bind dynamically multiple times per
- thread, just like a regular ``MetaData``.
-
- """
-
- __visit_name__ = "metadata"
-
- def __init__(self):
- """Construct a ThreadLocalMetaData."""
-
- self.context = util.threading.local()
- self.__engines = {}
- super(ThreadLocalMetaData, self).__init__()
-
- def bind(self):
- """The bound Engine or Connection for this thread.
-
- This property may be assigned an Engine or Connection, or assigned a
- string or URL to automatically create a basic Engine for this bind
- with ``create_engine()``."""
-
- return getattr(self.context, "_engine", None)
-
- @util.preload_module("sqlalchemy.engine.url")
- def _bind_to(self, bind):
- """Bind to a Connectable in the caller's thread."""
- url = util.preloaded.engine_url
- if isinstance(bind, (str, url.URL)):
- try:
- self.context._engine = self.__engines[bind]
- except KeyError:
- e = sqlalchemy.create_engine(bind)
- self.__engines[bind] = e
- self.context._engine = e
- else:
- # TODO: this is squirrely. we shouldn't have to hold onto engines
- # in a case like this
- if bind not in self.__engines:
- self.__engines[bind] = bind
- self.context._engine = bind
-
- bind = property(bind, _bind_to)
-
- def is_bound(self):
- """True if there is a bind for this thread."""
- return (
- hasattr(self.context, "_engine")
- and self.context._engine is not None
- )
-
- def dispose(self):
- """Dispose all bound engines, in all thread contexts."""
-
- for e in self.__engines.values():
- if hasattr(e, "dispose"):
- e.dispose()
-
-
class Computed(FetchedValue, SchemaItem):
"""Defines a generated column, i.e. "GENERATED ALWAYS AS" syntax.
self, collist, **kwargs
).select_from(self)
- @property
- @util.deprecated_20(
- ":attr:`.Executable.bind`",
- alternative="Bound metadata is being removed as of SQLAlchemy 2.0.",
- enable_warnings=False,
- )
- def bind(self):
- """Return the bound engine associated with either the left or right
- side of this :class:`_sql.Join`.
-
- """
-
- return self.left.bind or self.right.bind
-
@util.preload_module("sqlalchemy.sql.util")
def _anonymous_fromclause(self, name=None, flat=False):
sqlutil = util.preloaded.sql_util
def _from_objects(self):
return [self]
- @property
- def bind(self):
- return self.element.bind
-
class Alias(roles.DMLTableRole, AliasedReturnsRows):
"""Represents an table or selectable alias (AS).
_fetch_clause_options = None
_for_update_arg = None
- @util.deprecated_params(
- bind=(
- "2.0",
- "The :paramref:`_sql.select.bind` argument is deprecated and "
- "will be removed in SQLAlchemy 2.0.",
- ),
- )
def __init__(
self,
_label_style=LABEL_STYLE_DEFAULT,
offset=None,
order_by=None,
group_by=None,
- bind=None,
):
if use_labels:
if util.SQLALCHEMY_WARN_20:
if group_by is not None:
self.group_by.non_generative(self, *util.to_list(group_by))
- self._bind = bind
-
@_generative
def with_for_update(
self,
"""
return self.selects[0].selected_columns
- @property
- @util.deprecated_20(
- ":attr:`.Executable.bind`",
- alternative="Bound metadata is being removed as of SQLAlchemy 2.0.",
- enable_warnings=False,
- )
- def bind(self):
- """Returns the :class:`_engine.Engine` or :class:`_engine.Connection`
- to which this :class:`.Executable` is bound, or None if none found.
-
- """
- if self._bind:
- return self._bind
- for s in self.selects:
- e = s.bind
- if e:
- return e
- else:
- return None
-
- @bind.setter
- def bind(self, bind):
- self._bind = bind
-
class DeprecatedSelectGenerations:
"""A collection of methods available on :class:`_sql.Select`, these
- full description of explicit
FROM clause specification.
- :param bind=None:
- an :class:`_engine.Engine` or :class:`_engine.Connection` instance
- to which the
- resulting :class:`_expression.Select` object will be bound. The
- :class:`_expression.Select`
- object will otherwise automatically bind to
- whatever :class:`~.base.Connectable` instances can be located within
- its contained :class:`_expression.ClauseElement` members.
-
:param correlate=True:
indicates that this :class:`_expression.Select`
object should have its
"""
return CompoundSelect._create_intersect_all(self, *other, **kwargs)
- @property
- @util.deprecated_20(
- ":attr:`.Executable.bind`",
- alternative="Bound metadata is being removed as of SQLAlchemy 2.0.",
- enable_warnings=False,
- )
- def bind(self):
- """Returns the :class:`_engine.Engine` or :class:`_engine.Connection`
- to which this :class:`.Executable` is bound, or None if none found.
-
- """
- if self._bind:
- return self._bind
-
- for item in self._iterate_from_elements():
- if item._is_subquery and item.element is self:
- raise exc.InvalidRequestError(
- "select() construct refers to itself as a FROM"
- )
-
- e = item.bind
- if e:
- self._bind = e
- return e
- else:
- break
-
- for c in self._raw_columns:
- e = c.bind
- if e:
- self._bind = e
- return e
-
- @bind.setter
- def bind(self, bind):
- self._bind = bind
-
class ScalarSelect(roles.InElementRole, Generative, Grouping):
"""Represent a scalar subquery.
def _ensure_disambiguated_names(self):
return self
- @property
- def _bind(self):
- return self.element._bind
-
@_generative
def bindparams(self, *binds, **bind_as_values):
self.element = self.element.bindparams(*binds, **bind_as_values)
from . import operators
from . import roles
from . import type_api
-from .base import _bind_or_error
from .base import NO_ARG
from .base import SchemaEventTarget
from .elements import _NONE_NAME
**kw
)
- @property
- def bind(self):
- return self.metadata and self.metadata.bind or None
-
- def create(self, bind=None, checkfirst=False):
+ def create(self, bind, checkfirst=False):
"""Issue CREATE DDL for this type, if applicable."""
- if bind is None:
- bind = _bind_or_error(self)
t = self.dialect_impl(bind.dialect)
if t.__class__ is not self.__class__ and isinstance(t, SchemaType):
- t.create(bind=bind, checkfirst=checkfirst)
+ t.create(bind, checkfirst=checkfirst)
- def drop(self, bind=None, checkfirst=False):
+ def drop(self, bind, checkfirst=False):
"""Issue DROP DDL for this type, if applicable."""
- if bind is None:
- bind = _bind_or_error(self)
t = self.dialect_impl(bind.dialect)
if t.__class__ is not self.__class__ and isinstance(t, SchemaType):
- t.drop(bind=bind, checkfirst=checkfirst)
+ t.drop(bind, checkfirst=checkfirst)
def _on_table_create(self, target, bind, **kw):
if not self._is_impl_for_variant(bind.dialect, kw):
from unittest.mock import Mock
import sqlalchemy as tsa
-import sqlalchemy as sa
from sqlalchemy import create_engine
-from sqlalchemy import engine
from sqlalchemy import event
from sqlalchemy import exc
from sqlalchemy import ForeignKey
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
-from sqlalchemy import text
-from sqlalchemy import ThreadLocalMetaData
from sqlalchemy.engine import BindTyping
from sqlalchemy.engine import reflection
from sqlalchemy.engine.base import Connection
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
-from sqlalchemy.testing import is_false
from sqlalchemy.testing import is_instance_of
-from sqlalchemy.testing import is_true
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import expect_deprecated
from sqlalchemy.testing.assertions import expect_raises_message
with inspector._operation_context() as conn:
is_instance_of(conn, Connection)
- def test_bind_close_engine(self):
- e = testing.db
- with e.connect() as conn:
- assert not conn.closed
- assert conn.closed
-
- def test_bind_create_drop_err_metadata(self):
- metadata = MetaData()
- Table("test_table", metadata, Column("foo", Integer))
- for meth in [metadata.create_all, metadata.drop_all]:
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- assert_raises_message(
- exc.UnboundExecutionError,
- "MetaData object is not bound to an Engine or Connection.",
- meth,
- )
-
- def test_bind_create_drop_err_table(self):
- metadata = MetaData()
- table = Table("test_table", metadata, Column("foo", Integer))
-
- for meth in [table.create, table.drop]:
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- assert_raises_message(
- exc.UnboundExecutionError,
- (
- "Table object 'test_table' is not bound to an "
- "Engine or Connection."
- ),
- meth,
- )
-
- def test_bind_create_drop_bound(self):
-
- for meta in (MetaData, ThreadLocalMetaData):
- for bind in (testing.db, testing.db.connect()):
- if isinstance(bind, engine.Connection):
- bind.begin()
-
- if meta is ThreadLocalMetaData:
- with testing.expect_deprecated(
- "ThreadLocalMetaData is deprecated"
- ):
- metadata = meta()
- else:
- metadata = meta()
- table = Table("test_table", metadata, Column("foo", Integer))
- metadata.bind = bind
- assert metadata.bind is table.bind is bind
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- metadata.create_all()
-
- with testing.expect_deprecated(
- r"The Table.exists\(\) method is deprecated and will "
- "be removed in a future release."
- ):
- assert table.exists()
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- metadata.drop_all()
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- table.create()
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- table.drop()
- with testing.expect_deprecated(
- r"The Table.exists\(\) method is deprecated and will "
- "be removed in a future release."
- ):
- assert not table.exists()
-
- if meta is ThreadLocalMetaData:
- with testing.expect_deprecated(
- "ThreadLocalMetaData is deprecated"
- ):
- metadata = meta()
- else:
- metadata = meta()
-
- table = Table("test_table", metadata, Column("foo", Integer))
-
- metadata.bind = bind
-
- assert metadata.bind is table.bind is bind
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- metadata.create_all()
- with testing.expect_deprecated(
- r"The Table.exists\(\) method is deprecated and will "
- "be removed in a future release."
- ):
- assert table.exists()
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- metadata.drop_all()
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- table.create()
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL"
- ):
- table.drop()
- with testing.expect_deprecated(
- r"The Table.exists\(\) method is deprecated and will "
- "be removed in a future release."
- ):
- assert not table.exists()
- if isinstance(bind, engine.Connection):
- bind.close()
-
- def test_bind_create_drop_constructor_bound(self):
- for bind in (testing.db, testing.db.connect()):
- if isinstance(bind, engine.Connection):
- bind.begin()
- try:
- for args in (([bind], {}), ([], {"bind": bind})):
- with testing.expect_deprecated_20(
- "The MetaData.bind argument is deprecated "
- ):
- metadata = MetaData(*args[0], **args[1])
- table = Table(
- "test_table", metadata, Column("foo", Integer)
- )
- assert metadata.bind is table.bind is bind
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods "
- "that invoke SQL"
- ):
- metadata.create_all()
- is_true(inspect(bind).has_table(table.name))
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods "
- "that invoke SQL"
- ):
- metadata.drop_all()
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods "
- "that invoke SQL"
- ):
- table.create()
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods "
- "that invoke SQL"
- ):
- table.drop()
- is_false(inspect(bind).has_table(table.name))
- finally:
- if isinstance(bind, engine.Connection):
- bind.close()
-
- def test_bind_clauseelement(self, metadata):
- table = Table("test_table", metadata, Column("foo", Integer))
- metadata.create_all(bind=testing.db)
- for elem in [
- table.select,
- lambda **kwargs: sa.func.current_timestamp(**kwargs).select(),
- # func.current_timestamp().select,
- lambda **kwargs: text("select * from test_table", **kwargs),
- ]:
- with testing.db.connect() as bind:
- with testing.expect_deprecated_20(
- "The .*bind argument is deprecated"
- ):
- e = elem(bind=bind)
- assert e.bind is bind
-
def test_inspector_constructor_engine(self):
with testing.expect_deprecated(
r"The __init__\(\) method on Inspector is deprecated and will "
+++ /dev/null
-from sqlalchemy import Integer
-from sqlalchemy import testing
-from sqlalchemy.orm import declarative_base
-from sqlalchemy.orm import registry
-from sqlalchemy.orm import Session
-from sqlalchemy.testing import fixtures
-from sqlalchemy.testing import is_
-from sqlalchemy.testing.schema import Column
-
-
-class BoundMetadataDeclarativeTest(fixtures.MappedTest):
- def test_bound_declarative_base(self):
- with testing.expect_deprecated(
- "The ``bind`` argument to declarative_base"
- ):
- Base = declarative_base(testing.db)
-
- class User(Base):
- __tablename__ = "user"
- id = Column(Integer, primary_key=True)
-
- s = Session()
-
- with testing.expect_deprecated_20(
- "This Session located a target engine via bound metadata"
- ):
- is_(s.get_bind(User), testing.db)
-
- def test_bound_cls_registry_base(self):
- reg = registry(_bind=testing.db)
-
- Base = reg.generate_base()
-
- class User(Base):
- __tablename__ = "user"
- id = Column(Integer, primary_key=True)
-
- s = Session()
- with testing.expect_deprecated_20(
- "This Session located a target engine via bound metadata"
- ):
- is_(s.get_bind(User), testing.db)
-
- def test_bound_cls_registry_decorated(self):
- reg = registry(_bind=testing.db)
-
- @reg.mapped
- class User:
- __tablename__ = "user"
- id = Column(Integer, primary_key=True)
-
- s = Session()
-
- with testing.expect_deprecated_20(
- "This Session located a target engine via bound metadata"
- ):
- is_(s.get_bind(User), testing.db)
from sqlalchemy.engine import default
from sqlalchemy.engine import result_tuple
from sqlalchemy.orm import aliased
-from sqlalchemy.orm import as_declarative
from sqlalchemy.orm import attributes
from sqlalchemy.orm import backref
from sqlalchemy.orm import clear_mappers
from sqlalchemy.orm import contains_alias
from sqlalchemy.orm import contains_eager
from sqlalchemy.orm import declarative_base
-from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import defaultload
from sqlalchemy.orm import defer
from sqlalchemy.orm import deferred
from .inheritance._poly_fixtures import Manager
from .inheritance._poly_fixtures import Person
from .test_ac_relationships import PartitionByFixture
-from .test_bind import GetBindTest as _GetBindTest
-from .test_default_strategies import (
- DefaultStrategyOptionsTest as _DefaultStrategyOptionsTest,
-)
from .test_deferred import InheritanceTest as _deferred_InheritanceTest
from .test_dynamic import _DynamicFixture
from .test_events import _RemoveListeners
from .test_transaction import _LocalFixture
from ..sql.test_compare import CacheKeyFixture
+if True:
+ # hack - zimports won't stop reformatting this to be too-long for now
+ from .test_default_strategies import (
+ DefaultStrategyOptionsTest as _DefaultStrategyOptionsTest,
+ )
join_aliased_dep = (
r"The ``aliased`` and ``from_joinpoint`` keyword arguments to "
eq_(list(q2), [(True,), (False,), (False,), (False,)])
-class DeclarativeBind(fixtures.TestBase):
- def test_declarative_base(self):
- with testing.expect_deprecated_20(
- "The ``bind`` argument to declarative_base is "
- "deprecated and will be removed in SQLAlchemy 2.0.",
- ):
- Base = declarative_base(bind=testing.db)
-
- is_true(Base.metadata.bind is testing.db)
-
- def test_as_declarative(self):
- with testing.expect_deprecated_20(
- "The ``bind`` argument to as_declarative is "
- "deprecated and will be removed in SQLAlchemy 2.0.",
- ):
-
- @as_declarative(bind=testing.db)
- class Base:
- @declared_attr
- def __tablename__(cls):
- return cls.__name__.lower()
-
- id = Column(Integer, primary_key=True)
-
- is_true(Base.metadata.bind is testing.db)
-
-
class JoinTest(QueryTest, AssertsCompiledSQL):
__dialect__ = "default"
self._test(False, True, False)
-class GetBindTest(_GetBindTest):
- @classmethod
- def define_tables(cls, metadata):
- super(GetBindTest, cls).define_tables(metadata)
- metadata.bind = testing.db
-
- def test_fallback_table_metadata(self):
- session = self._fixture({})
- with testing.expect_deprecated_20(
- "This Session located a target engine via bound metadata"
- ):
- is_(session.get_bind(self.classes.BaseClass), testing.db)
-
- def test_bind_base_table_concrete_sub_class(self):
- base_class_bind = Mock()
- session = self._fixture({self.tables.base_table: base_class_bind})
-
- with testing.expect_deprecated_20(
- "This Session located a target engine via bound metadata"
- ):
- is_(session.get_bind(self.classes.ConcreteSubClass), testing.db)
-
-
class DeprecationScopedSessionTest(fixtures.MappedTest):
def test_config_errors(self):
sm = sessionmaker()
from sqlalchemy import exists
from sqlalchemy import ForeignKey
from sqlalchemy import func
-from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import join
from sqlalchemy import literal_column
eq_(t2.name, "t")
-class BoundMetadataTest(fixtures.TestBase):
- def test_arg_deprecated(self):
- with testing.expect_deprecated_20(
- "The MetaData.bind argument is deprecated"
- ):
- m1 = MetaData(testing.db)
-
- Table("t", m1, Column("q", Integer))
-
- with testing.expect_deprecated_20(
- "The ``bind`` argument for schema methods that invoke SQL "
- "against an engine or connection will be required"
- ):
- m1.create_all()
- try:
- assert "t" in inspect(testing.db).get_table_names()
- finally:
- m1.drop_all(testing.db)
-
- assert "t" not in inspect(testing.db).get_table_names()
-
- def test_bind_arg_text(self):
- with testing.expect_deprecated_20(
- "The text.bind argument is deprecated and will be "
- "removed in SQLAlchemy 2.0."
- ):
- t1 = text("ASdf", bind=testing.db)
-
- # no warnings emitted
- is_(t1.bind, testing.db)
- eq_(str(t1), "ASdf")
-
- def test_bind_arg_function(self):
- with testing.expect_deprecated_20(
- "The text.bind argument is deprecated and will be "
- "removed in SQLAlchemy 2.0."
- ):
- f1 = func.foobar(bind=testing.db)
-
- # no warnings emitted
- is_(f1.bind, testing.db)
- eq_(str(f1), "foobar()")
-
- def test_bind_arg_select(self):
- with testing.expect_deprecated_20(
- "The select.bind argument is deprecated and will be "
- "removed in SQLAlchemy 2.0."
- ):
- s1 = select([column("q")], bind=testing.db)
-
- # no warnings emitted
- is_(s1.bind, testing.db)
- eq_(str(s1), "SELECT q")
-
- def test_bind_attr_join_no_warning(self):
- t1 = table("t1", column("a"))
- t2 = table("t2", column("b"))
- j1 = join(t1, t2, t1.c.a == t2.c.b)
-
- # no warnings emitted
- is_(j1.bind, None)
- eq_(str(j1), "t1 JOIN t2 ON t1.a = t2.b")
-
-
class DeprecationWarningsTest(fixtures.TestBase, AssertsCompiledSQL):
__backend__ = True
assert _op_modern == _op_legacy
-class DDLDeprecatedBindTest(fixtures.TestBase):
- def teardown_test(self):
- with testing.db.begin() as conn:
- if inspect(conn).has_table("foo"):
- conn.execute(schema.DropTable(table("foo")))
-
- @testing.combinations(
- (schema.AddConstraint,),
- (schema.DropConstraint,),
- (schema.CreateSequence,),
- (schema.DropSequence,),
- (schema.CreateSchema,),
- (schema.DropSchema,),
- (schema.SetTableComment,),
- (schema.DropTableComment,),
- (schema.SetColumnComment,),
- (schema.DropColumnComment,),
- )
- def test_bind_other_constructs(self, const):
- m1 = mock.Mock()
-
- with testing.expect_deprecated_20(
- "The DDLElement.bind argument is deprecated"
- ):
- c1 = const(m1, bind=testing.db)
-
- is_(c1.bind, testing.db)
-
-
class FutureSelectTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
def test_pickle():
meta.bind = testing.db
meta2 = pickle.loads(pickle.dumps(meta))
- assert meta2.bind is None
pickle.loads(pickle.dumps(meta2))
return (
meta2.tables["mytable"],
Table("othertable", meta2, autoload_with=testing.db)
Table("has_comments", meta2, autoload_with=testing.db)
meta3 = pickle.loads(pickle.dumps(meta2))
- assert meta3.bind is None
assert meta3.tables["mytable"] is not t1
return (
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
