.. changelog::
:version: 0.9.2
+ .. change::
+ :tags: feature, sql
+ :tickets: 2866
+
+ The system by which schema constructs and certain SQL constructs
+ accept dialect-specific keyword arguments has been enhanced. This
+ system includes commonly the :class:`.Table` and :class:`.Index` constructs,
+ which accept a wide variety of dialect-specific arguments such as
+ ``mysql_engine`` and ``postgresql_where``, as well as the constructs
+ :class:`.PrimaryKeyConstraint`, :class:`.UniqueConstraint`,
+ :class:`.Update`, :class:`.Insert` and :class:`.Delete`, and also
+ newly added kwarg capability to :class:`.ForeignKeyConstraint`
+ and :class:`.ForeignKey`. The change is that participating dialects
+ can now specify acceptable argument lists for these constructs, allowing
+ an argument error to be raised if an invalid keyword is specified for
+ a particular dialect. If the dialect portion of the keyword is unrecognized,
+ a warning is emitted only; while the system will actually make use
+ of setuptools entrypoints in order to locate non-local dialects,
+ the use case where certain dialect-specific arguments are used
+ in an environment where that third-party dialect is uninstalled remains
+ supported. Dialects also have to explicitly opt-in to this system,
+ so that external dialects which aren't making use of this system
+ will remain unaffected.
+
.. change::
:tags: bug, sql
:pullreq: bitbucket:11
colspecs = colspecs
ischema_names = ischema_names
+ construct_arguments = []
+
# defaults to dialect ver. 3,
# will be autodetected off upon
# first connect
text += "UNIQUE "
# handle clustering option
- if index.kwargs.get("mssql_clustered"):
+ if index.dialect_options['mssql']['clustered']:
text += "CLUSTERED "
text += "INDEX %s ON %s (%s)" \
)
# handle other included columns
- if index.kwargs.get("mssql_include"):
+ if index.dialect_options['mssql']['include']:
inclusions = [index.table.c[col]
if isinstance(col, util.string_types) else col
- for col in index.kwargs["mssql_include"]]
+ for col in index.dialect_options['mssql']['include']]
text += " INCLUDE (%s)" \
% ', '.join([preparer.quote(c.name)
self.preparer.format_constraint(constraint)
text += "PRIMARY KEY "
- # support clustered option
- if constraint.kwargs.get("mssql_clustered"):
+ if constraint.dialect_options['mssql']['clustered']:
text += "CLUSTERED "
text += "(%s)" % ', '.join(self.preparer.quote(c.name)
self.preparer.format_constraint(constraint)
text += "UNIQUE "
- # support clustered option
- if constraint.kwargs.get("mssql_clustered"):
+ if constraint.dialect_options['mssql']['clustered']:
text += "CLUSTERED "
text += "(%s)" % ', '.join(self.preparer.quote(c.name)
type_compiler = MSTypeCompiler
preparer = MSIdentifierPreparer
+ construct_arguments = [
+ (sa_schema.PrimaryKeyConstraint, {
+ "clustered": False
+ }),
+ (sa_schema.UniqueConstraint, {
+ "clustered": False
+ }),
+ (sa_schema.Index, {
+ "clustered": False,
+ "include": None
+ })
+ ]
+
def __init__(self,
query_timeout=None,
use_scope_identity=True,
constraint_string = super(
MySQLDDLCompiler, self).create_table_constraints(table)
- engine_key = '%s_engine' % self.dialect.name
- is_innodb = engine_key in table.kwargs and \
- table.kwargs[engine_key].lower() == 'innodb'
+ # why self.dialect.name and not 'mysql'? because of drizzle
+ is_innodb = 'engine' in table.dialect_options[self.dialect.name] and \
+ table.dialect_options[self.dialect.name]['engine'].lower() == 'innodb'
auto_inc_column = table._autoincrement_column
text += "UNIQUE "
text += "INDEX %s ON %s " % (name, table)
- if 'mysql_length' in index.kwargs:
- length = index.kwargs['mysql_length']
+ length = index.dialect_options['mysql']['length']
+ if length is not None:
if isinstance(length, dict):
# length value can be a (column_name --> integer value) mapping
columns = ', '.join(columns)
text += '(%s)' % columns
- if 'mysql_using' in index.kwargs:
- using = index.kwargs['mysql_using']
+ using = index.dialect_options['mysql']['using']
+ if using is not None:
text += " USING %s" % (preparer.quote(using))
return text
def visit_primary_key_constraint(self, constraint):
text = super(MySQLDDLCompiler, self).\
visit_primary_key_constraint(constraint)
- if "mysql_using" in constraint.kwargs:
- using = constraint.kwargs['mysql_using']
+ using = constraint.dialect_options['mysql']['using']
+ if using:
text += " USING %s" % (self.preparer.quote(using))
return text
_backslash_escapes = True
_server_ansiquotes = False
+ construct_arguments = [
+ (sa_schema.Table, {
+ "*": None
+ }),
+ (sql.Update, {
+ "limit": None
+ }),
+ (sa_schema.PrimaryKeyConstraint, {
+ "using": None
+ }),
+ (sa_schema.Index, {
+ "using": None,
+ "length": None,
+ })
+ ]
+
def __init__(self, isolation_level=None, **kwargs):
kwargs.pop('use_ansiquotes', None) # legacy
default.DefaultDialect.__init__(self, **kwargs)
reflection_options = ('oracle_resolve_synonyms', )
+ construct_arguments = []
+
def __init__(self,
use_ansi=True,
optimize_limits=False,
preparer.format_table(index.table)
)
- if 'postgresql_using' in index.kwargs:
- using = index.kwargs['postgresql_using']
+ using = index.dialect_options['postgresql']['using']
+ if using:
text += "USING %s " % preparer.quote(using)
- ops = index.kwargs.get('postgresql_ops', {})
+ ops = index.dialect_options["postgresql"]["ops"]
text += "(%s)" \
% (
', '.join([
for expr, c in zip(index.expressions, index.columns)])
)
- if 'postgresql_where' in index.kwargs:
- whereclause = index.kwargs['postgresql_where']
- else:
- whereclause = None
+ whereclause = index.dialect_options["postgresql"]["where"]
if whereclause is not None:
where_compiled = self.sql_compiler.process(
inspector = PGInspector
isolation_level = None
+ construct_arguments = [
+ (schema.Index, {
+ "using": False,
+ "where": None,
+ "ops": {}
+ })
+ ]
+
_backslash_escapes = True
def __init__(self, isolation_level=None, json_serializer=None,
import datetime
import re
-from sqlalchemy import sql, exc
-from sqlalchemy.engine import default, base, reflection
-from sqlalchemy import types as sqltypes
-from sqlalchemy import util
-from sqlalchemy.sql import compiler
-from sqlalchemy import processors
-
-from sqlalchemy.types import BIGINT, BLOB, BOOLEAN, CHAR,\
+from ... import sql, exc
+from ...engine import default, reflection
+from ... import types as sqltypes, schema as sa_schema
+from ... import util
+from ...sql import compiler
+from ... import processors
+
+from ...types import BIGINT, BLOB, BOOLEAN, CHAR,\
DECIMAL, FLOAT, REAL, INTEGER, NUMERIC, SMALLINT, TEXT,\
TIMESTAMP, VARCHAR
colspec += " NOT NULL"
if (column.primary_key and
- column.table.kwargs.get('sqlite_autoincrement', False) and
+ column.table.dialect_options['sqlite']['autoincrement'] and
len(column.table.primary_key.columns) == 1 and
issubclass(column.type._type_affinity, sqltypes.Integer) and
not column.foreign_keys):
if len(constraint.columns) == 1:
c = list(constraint)[0]
if c.primary_key and \
- c.table.kwargs.get('sqlite_autoincrement', False) and \
+ c.table.dialect_options['sqlite']['autoincrement'] and \
issubclass(c.type._type_affinity, sqltypes.Integer) and \
not c.foreign_keys:
return None
supports_cast = True
supports_default_values = True
+ construct_arguments = [
+ (sa_schema.Table, {
+ "autoincrement": False
+ })
+ ]
+
_broken_fk_pragma_quotes = False
def __init__(self, isolation_level=None, native_datetime=False, **kwargs):
preparer = SybaseIdentifierPreparer
inspector = SybaseInspector
+ construct_arguments = []
+
def _get_default_schema_name(self, connection):
return connection.scalar(
text("SELECT user_name() as user_name",
server_version_info = None
+ construct_arguments = None
+ """Optional set of argument specifiers for various SQLAlchemy
+ constructs, typically schema items.
+
+ To
+ implement, establish as a series of tuples, as in::
+
+ construct_arguments = [
+ (schema.Index, {
+ "using": False,
+ "where": None,
+ "ops": None
+ })
+ ]
+
+ If the above construct is established on the Postgresql dialect,
+ the ``Index`` construct will now accept additional keyword arguments
+ such as ``postgresql_using``, ``postgresql_where``, etc. Any kind of
+ ``postgresql_XYZ`` argument not corresponding to the above template will
+ be rejected with an ``ArgumentError`, for all those SQLAlchemy constructs
+ which implement the :class:`.DialectKWArgs` class.
+
+ The default is ``None``; older dialects which don't implement the argument
+ will have the old behavior of un-validated kwargs to schema/SQL constructs.
+
+ """
+
# indicates symbol names are
# UPPERCASEd if they are case insensitive
# within the database.
self._decoder = processors.to_unicode_processor_factory(self.encoding)
+
@util.memoized_property
def _type_memos(self):
return weakref.WeakKeyDictionary()
# apply table options
tbl_opts = self.get_table_options(table_name, schema, **table.kwargs)
if tbl_opts:
- table.kwargs.update(tbl_opts)
+ table._validate_dialect_kwargs(tbl_opts)
# table.kwargs will need to be passed to each reflection method. Make
# sure keywords are strings.
"""
+class NoSuchModuleError(ArgumentError):
+ """Raised when a dynamically-loaded module (usually a database dialect)
+ of a particular name cannot be located."""
class NoForeignKeysError(ArgumentError):
"""Raised when no foreign keys can be located between two selectables
from .. import util, exc
import itertools
from .visitors import ClauseVisitor
-
+import re
+import collections
PARSE_AUTOCOMMIT = util.symbol('PARSE_AUTOCOMMIT')
NO_ARG = util.symbol('NO_ARG')
return self
+class DialectKWArgs(object):
+ """Establish the ability for a class to have dialect-specific arguments
+ with defaults and validation.
+
+ """
+
+ @util.memoized_property
+ def dialect_kwargs(self):
+ """A collection of keyword arguments specified as dialect-specific
+ options to this construct.
+
+ The arguments are present here in their original ``<dialect>_<kwarg>``
+ format.
+
+ .. versionadded:: 0.9.2
+
+ .. seealso::
+
+ :attr:`.DialectKWArgs.dialect_options` - nested dictionary form
+
+ """
+
+ return util.immutabledict(
+ (
+ "%s_%s" % (dialect_name, kwarg_name),
+ kw_dict[kwarg_name]
+ )
+ for dialect_name, kw_dict in self.dialect_options.items()
+ for kwarg_name in kw_dict if kwarg_name != '*'
+ )
+
+ @property
+ def kwargs(self):
+ """Deprecated; see :attr:`.DialectKWArgs.dialect_kwargs"""
+ return self.dialect_kwargs
+
+ @util.dependencies("sqlalchemy.dialects")
+ def _kw_reg_for_dialect(dialects, dialect_name):
+ dialect_cls = dialects.registry.load(dialect_name)
+ if dialect_cls.construct_arguments is None:
+ return None
+ return dict(dialect_cls.construct_arguments)
+ _kw_registry = util.PopulateDict(_kw_reg_for_dialect)
+
+ def _kw_reg_for_dialect_cls(self, dialect_name):
+ construct_arg_dictionary = DialectKWArgs._kw_registry[dialect_name]
+ if construct_arg_dictionary is None:
+ return {"*": None}
+ else:
+ d = {}
+ for cls in reversed(self.__class__.__mro__):
+ if cls in construct_arg_dictionary:
+ d.update(construct_arg_dictionary[cls])
+ return d
+
+ @util.memoized_property
+ def dialect_options(self):
+ """A collection of keyword arguments specified as dialect-specific
+ options to this construct.
+
+ This is a two-level nested registry, keyed to ``<dialect_name>``
+ and ``<argument_name>``. For example, the ``postgresql_where`` argument
+ would be locatable as::
+
+ arg = my_object.dialect_options['postgresql']['where']
+
+ .. versionadded:: 0.9.2
+
+ .. seealso::
+
+ :attr:`.DialectKWArgs.dialect_kwargs` - flat dictionary form
+
+ """
+
+ return util.PopulateDict(
+ util.portable_instancemethod(self._kw_reg_for_dialect_cls)
+ )
+
+ def _validate_dialect_kwargs(self, kwargs):
+ # validate remaining kwargs that they all specify DB prefixes
+
+ if not kwargs:
+ return
+
+ self.__dict__.pop('dialect_kwargs', None)
+
+ for k in kwargs:
+ m = re.match('^(.+?)_(.+)$', k)
+ if m is None:
+ raise TypeError("Additional arguments should be "
+ "named <dialectname>_<argument>, got '%s'" % k)
+ dialect_name, arg_name = m.group(1, 2)
+
+ try:
+ construct_arg_dictionary = self.dialect_options[dialect_name]
+ except exc.NoSuchModuleError:
+ util.warn(
+ "Can't validate argument %r; can't "
+ "locate any SQLAlchemy dialect named %r" %
+ (k, dialect_name))
+ self.dialect_options[dialect_name] = {
+ "*": None,
+ arg_name: kwargs[k]}
+ else:
+ if "*" not in construct_arg_dictionary and \
+ arg_name not in construct_arg_dictionary:
+ raise exc.ArgumentError(
+ "Argument %r is not accepted by "
+ "dialect %r on behalf of %r" % (
+ k,
+ dialect_name, self.__class__
+ ))
+ else:
+ construct_arg_dictionary[arg_name] = kwargs[k]
+
+
class Generative(object):
"""Allow a ClauseElement to generate itself via the
@_generative decorator.
"""
-from .base import Executable, _generative, _from_objects
+from .base import Executable, _generative, _from_objects, DialectKWArgs
from .elements import ClauseElement, _literal_as_text, Null, and_, _clone
from .selectable import _interpret_as_from, _interpret_as_select, HasPrefixes
from .. import util
from .. import exc
-class UpdateBase(HasPrefixes, Executable, ClauseElement):
+class UpdateBase(DialectKWArgs, HasPrefixes, Executable, ClauseElement):
"""Form the base for ``INSERT``, ``UPDATE``, and ``DELETE`` statements.
"""
_execution_options = \
Executable._execution_options.union({'autocommit': True})
- kwargs = util.immutabledict()
_hints = util.immutabledict()
_prefixes = ()
prefixes=None,
returning=None,
return_defaults=False,
- **kwargs):
+ **dialect_kw):
"""Construct an :class:`.Insert` object.
Similar functionality is available via the
self.select = self.select_names = None
self.inline = inline
self._returning = returning
- self.kwargs = kwargs
+ self._validate_dialect_kwargs(dialect_kw)
self._return_defaults = return_defaults
def get_children(self, **kwargs):
prefixes=None,
returning=None,
return_defaults=False,
- **kwargs):
+ **dialect_kw):
"""Construct an :class:`.Update` object.
E.g.::
else:
self._whereclause = None
self.inline = inline
- self.kwargs = kwargs
+ self._validate_dialect_kwargs(dialect_kw)
self._return_defaults = return_defaults
bind=None,
returning=None,
prefixes=None,
- **kwargs):
+ **dialect_kw):
"""Construct :class:`.Delete` object.
Similar functionality is available via the
else:
self._whereclause = None
- self.kwargs = kwargs
+ self._validate_dialect_kwargs(dialect_kw)
def get_children(self, **kwargs):
if self._whereclause is not None:
"""
-import re
import inspect
from .. import exc, util, event, inspection
-from .base import SchemaEventTarget
+from .base import SchemaEventTarget, DialectKWArgs
from . import visitors
from . import type_api
from .base import _bind_or_error, ColumnCollection
return schema + "." + name
-def _validate_dialect_kwargs(kwargs, name):
- # validate remaining kwargs that they all specify DB prefixes
-
- for k in kwargs:
- m = re.match('^(.+?)_.*', k)
- if m is None:
- raise TypeError("Additional arguments should be "
- "named <dialectname>_<argument>, got '%s'" % k)
@inspection._self_inspects
class SchemaItem(SchemaEventTarget, visitors.Visitable):
return schema_item
-class Table(SchemaItem, TableClause):
+class Table(DialectKWArgs, SchemaItem, TableClause):
"""Represent a table in a database.
e.g.::
``quote_schema=True`` to the constructor, or use the :class:`.quoted_name`
construct to specify the name.
-
:param useexisting: Deprecated. Use extend_existing.
+ :param \**kw: Additional keyword arguments not mentioned above are
+ dialect specific, and passed in the form ``<dialectname>_<argname>``.
+ See the documentation regarding an individual dialect at
+ :ref:`dialect_toplevel` for detail on documented arguments.
+
"""
__visit_name__ = 'table'
PrimaryKeyConstraint()._set_parent_with_dispatch(self)
self.foreign_keys = set()
self._extra_dependencies = set()
- self.kwargs = {}
if self.schema is not None:
self.fullname = "%s.%s" % (self.schema, self.name)
else:
self._init_items(*args)
def _extra_kwargs(self, **kwargs):
- # validate remaining kwargs that they all specify DB prefixes
- _validate_dialect_kwargs(kwargs, "Table")
- self.kwargs.update(kwargs)
+ self._validate_dialect_kwargs(kwargs)
def _init_collections(self):
pass
return ColumnClause.get_children(self, **kwargs)
-class ForeignKey(SchemaItem):
+class ForeignKey(DialectKWArgs, SchemaItem):
"""Defines a dependency between two columns.
``ForeignKey`` is specified as an argument to a :class:`.Column` object,
def __init__(self, column, _constraint=None, use_alter=False, name=None,
onupdate=None, ondelete=None, deferrable=None,
- initially=None, link_to_name=False, match=None):
+ initially=None, link_to_name=False, match=None,
+ **dialect_kw):
"""
Construct a column-level FOREIGN KEY.
DDL for this constraint. Typical values include SIMPLE, PARTIAL
and FULL.
+ :param \**dialect_kw: Additional keyword arguments are dialect specific,
+ and passed in the form ``<dialectname>_<argname>``. The arguments
+ are ultimately handled by a corresponding :class:`.ForeignKeyConstraint`.
+ See the documentation regarding an individual dialect at
+ :ref:`dialect_toplevel` for detail on documented arguments.
+
+ .. versionadded:: 0.9.2
+
"""
self._colspec = column
self.initially = initially
self.link_to_name = link_to_name
self.match = match
+ self._unvalidated_dialect_kw = dialect_kw
def __repr__(self):
return "ForeignKey(%r)" % self._get_colspec()
deferrable=self.deferrable,
initially=self.initially,
link_to_name=self.link_to_name,
- match=self.match
+ match=self.match,
+ **self._unvalidated_dialect_kw
)
return self._schema_item_copy(fk)
onupdate=self.onupdate, ondelete=self.ondelete,
deferrable=self.deferrable, initially=self.initially,
match=self.match,
+ **self._unvalidated_dialect_kw
)
self.constraint._elements[self.parent] = self
self.constraint._set_parent_with_dispatch(table)
DefaultClause.__init__(self, *arg, **kw)
-class Constraint(SchemaItem):
+class Constraint(DialectKWArgs, SchemaItem):
"""A table-level SQL constraint."""
__visit_name__ = 'constraint'
def __init__(self, name=None, deferrable=None, initially=None,
_create_rule=None,
- **kw):
+ **dialect_kw):
"""Create a SQL constraint.
:param name:
_create_rule is used by some types to create constraints.
Currently, its call signature is subject to change at any time.
- :param \**kwargs:
- Dialect-specific keyword parameters, see the documentation
- for various dialects and constraints regarding options here.
+ :param \**dialect_kw: Additional keyword arguments are dialect specific,
+ and passed in the form ``<dialectname>_<argname>``. See the
+ documentation regarding an individual dialect at :ref:`dialect_toplevel`
+ for detail on documented arguments.
"""
self.initially = initially
self._create_rule = _create_rule
util.set_creation_order(self)
- _validate_dialect_kwargs(kw, self.__class__.__name__)
- self.kwargs = kw
+ self._validate_dialect_kwargs(dialect_kw)
@property
def table(self):
Optional string. If set, emit INITIALLY <value> when issuing DDL
for this constraint.
+ :param \**kw: other keyword arguments including dialect-specific
+ arguments are propagated to the :class:`.Constraint` superclass.
+
"""
ColumnCollectionMixin.__init__(self, *columns)
Constraint.__init__(self, **kw)
def __init__(self, columns, refcolumns, name=None, onupdate=None,
ondelete=None, deferrable=None, initially=None, use_alter=False,
- link_to_name=False, match=None, table=None):
+ link_to_name=False, match=None, table=None, **dialect_kw):
"""Construct a composite-capable FOREIGN KEY.
:param columns: A sequence of local column names. The named columns
DDL for this constraint. Typical values include SIMPLE, PARTIAL
and FULL.
+ :param \**dialect_kw: Additional keyword arguments are dialect specific,
+ and passed in the form ``<dialectname>_<argname>``. See the
+ documentation regarding an individual dialect at :ref:`dialect_toplevel`
+ for detail on documented arguments.
+
+ .. versionadded:: 0.9.2
+
"""
super(ForeignKeyConstraint, self).\
- __init__(name, deferrable, initially)
+ __init__(name, deferrable, initially, **dialect_kw)
self.onupdate = onupdate
self.ondelete = ondelete
link_to_name=self.link_to_name,
match=self.match,
deferrable=self.deferrable,
- initially=self.initially
+ initially=self.initially,
+ **self.dialect_kwargs
)
if table is not None:
__visit_name__ = 'unique_constraint'
-class Index(ColumnCollectionMixin, SchemaItem):
+class Index(DialectKWArgs, ColumnCollectionMixin, SchemaItem):
"""A table-level INDEX.
Defines a composite (one or more column) INDEX.
be arbitrary SQL expressions which ultmately refer to a
:class:`.Column`.
- :param unique:
- Defaults to False: create a unique index.
+ :param unique=False:
+ Keyword only argument; if True, create a unique index.
+
+ :param quote=None:
+ Keyword only argument; whether to apply quoting to the name of
+ the index. Works in the same manner as that of
+ :paramref:`.Column.quote`.
- :param \**kw:
- Other keyword arguments may be interpreted by specific dialects.
+ :param \**kw: Additional keyword arguments not mentioned above are
+ dialect specific, and passed in the form ``<dialectname>_<argname>``.
+ See the documentation regarding an individual dialect at
+ :ref:`dialect_toplevel` for detail on documented arguments.
"""
self.table = None
self.expressions = expressions
self.name = quoted_name(name, kw.pop("quote", None))
self.unique = kw.pop('unique', False)
- self.kwargs = kw
+ self._validate_dialect_kwargs(kw)
# will call _set_parent() if table-bound column
# objects are present
self.impls[name] = impl.load
return impl.load()
- raise exc.ArgumentError(
+ raise exc.NoSuchModuleError(
"Can't load plugin: %s:%s" %
(self.group, name))
from sqlalchemy.testing import fixtures
from sqlalchemy import testing
from sqlalchemy.testing import ComparesTables, AssertsCompiledSQL
-from sqlalchemy.testing import eq_, is_
+from sqlalchemy.testing import eq_, is_, mock
+from contextlib import contextmanager
class MetaDataTest(fixtures.TestBase, ComparesTables):
def test_metadata_connect(self):
meta2 = MetaData()
table_c = table.tometadata(meta2)
+ eq_(table.kwargs, {"mysql_engine": "InnoDB"})
+
eq_(table.kwargs, table_c.kwargs)
def test_tometadata_indexes(self):
]
)
+class DialectKWArgTest(fixtures.TestBase):
+ @contextmanager
+ def _fixture(self):
+ from sqlalchemy.engine.default import DefaultDialect
+ class ParticipatingDialect(DefaultDialect):
+ construct_arguments = [
+ (schema.Index, {
+ "x": 5,
+ "y": False,
+ "z_one": None
+ }),
+ (schema.ForeignKeyConstraint, {
+ "foobar": False
+ })
+ ]
+
+ class ParticipatingDialect2(DefaultDialect):
+ construct_arguments = [
+ (schema.Index, {
+ "x": 9,
+ "y": True,
+ "pp": "default"
+ }),
+ (schema.Table, {
+ "*": None
+ })
+ ]
+
+ class NonParticipatingDialect(DefaultDialect):
+ construct_arguments = None
+
+ def load(dialect_name):
+ if dialect_name == "participating":
+ return ParticipatingDialect
+ elif dialect_name == "participating2":
+ return ParticipatingDialect2
+ elif dialect_name == "nonparticipating":
+ return NonParticipatingDialect
+ else:
+ raise exc.NoSuchModuleError("no dialect %r" % dialect_name)
+ with mock.patch("sqlalchemy.dialects.registry.load", load):
+ yield
+
+ def test_participating(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c', participating_y=True)
+ eq_(
+ idx.dialect_kwargs,
+ {
+ 'participating_z_one': None,
+ 'participating_y': True,
+ 'participating_x': 5
+ }
+ )
+
+ def test_nonparticipating(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c', nonparticipating_y=True, nonparticipating_q=5)
+ eq_(
+ idx.dialect_kwargs,
+ {
+ 'nonparticipating_y': True,
+ 'nonparticipating_q': 5
+ }
+ )
+
+ def test_unknown_dialect_warning(self):
+ with self._fixture():
+ assert_raises_message(
+ exc.SAWarning,
+ "Can't validate argument 'unknown_y'; can't locate "
+ "any SQLAlchemy dialect named 'unknown'",
+ Index, 'a', 'b', 'c', unknown_y=True
+ )
+
+ def test_participating_bad_kw(self):
+ with self._fixture():
+ assert_raises_message(
+ exc.ArgumentError,
+ "Argument 'participating_q_p_x' is not accepted by dialect "
+ "'participating' on behalf of "
+ "<class 'sqlalchemy.sql.schema.Index'>",
+ Index, 'a', 'b', 'c', participating_q_p_x=8
+ )
+
+ def test_participating_unknown_schema_item(self):
+ with self._fixture():
+ # the dialect doesn't include UniqueConstraint in
+ # its registry at all.
+ assert_raises_message(
+ exc.ArgumentError,
+ "Argument 'participating_q_p_x' is not accepted by dialect "
+ "'participating' on behalf of "
+ "<class 'sqlalchemy.sql.schema.UniqueConstraint'>",
+ UniqueConstraint, 'a', 'b', participating_q_p_x=8
+ )
+
+ @testing.emits_warning("Can't validate")
+ def test_unknown_dialect_warning_still_populates(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c', unknown_y=True)
+ eq_(idx.dialect_kwargs, {"unknown_y": True}) # still populates
+
+ @testing.emits_warning("Can't validate")
+ def test_unknown_dialect_warning_still_populates_multiple(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c', unknown_y=True, unknown_z=5,
+ otherunknown_foo='bar', participating_y=8)
+ eq_(idx.dialect_kwargs,
+ {'unknown_z': 5, 'participating_y': 8,
+ 'unknown_y': True, 'participating_z_one': None,
+ 'otherunknown_foo': 'bar', 'participating_x': 5}
+ ) # still populates
+
+ def test_combined(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c', participating_x=7,
+ nonparticipating_y=True)
+ eq_(
+ idx.dialect_kwargs,
+ {
+ 'participating_z_one': None,
+ 'participating_y': False,
+ 'participating_x': 7,
+ 'nonparticipating_y': True,
+ }
+ )
+
+ def test_multiple_participating(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c',
+ participating_x=7,
+ participating2_x=15,
+ participating2_y="lazy"
+ )
+ eq_(
+ idx.dialect_kwargs,
+ {
+ 'participating_z_one': None,
+ 'participating_x': 7,
+ 'participating_y': False,
+ 'participating2_pp': 'default',
+ 'participating2_x': 15,
+ 'participating2_y': 'lazy'
+ }
+ )
+
+ def test_foreign_key_propagate(self):
+ with self._fixture():
+ m = MetaData()
+ fk = ForeignKey('t2.id', participating_foobar=True)
+ t = Table('t', m, Column('id', Integer, fk))
+ fkc = [c for c in t.constraints if isinstance(c, ForeignKeyConstraint)][0]
+ eq_(
+ fkc.dialect_kwargs,
+ {'participating_foobar': True}
+ )
+
+ def test_foreign_key_propagate_exceptions_delayed(self):
+ with self._fixture():
+ m = MetaData()
+ fk = ForeignKey('t2.id', participating_fake=True)
+ c1 = Column('id', Integer, fk)
+ assert_raises_message(
+ exc.ArgumentError,
+ "Argument 'participating_fake' is not accepted by "
+ "dialect 'participating' on behalf of "
+ "<class 'sqlalchemy.sql.schema.ForeignKeyConstraint'>",
+ Table, 't', m, c1
+ )
+
+ def test_wildcard(self):
+ with self._fixture():
+ m = MetaData()
+ t = Table('x', m, Column('x', Integer),
+ participating2_xyz='foo',
+ participating2_engine='InnoDB',
+ )
+ eq_(
+ t.dialect_kwargs,
+ {
+ 'participating2_xyz': 'foo',
+ 'participating2_engine': 'InnoDB'
+ }
+ )
+
+ def test_uninit_wildcard(self):
+ with self._fixture():
+ m = MetaData()
+ t = Table('x', m, Column('x', Integer))
+ eq_(
+ t.dialect_options['participating2'], {'*': None}
+ )
+ eq_(
+ t.dialect_kwargs, {}
+ )
+
+ def test_not_contains_wildcard(self):
+ with self._fixture():
+ m = MetaData()
+ t = Table('x', m, Column('x', Integer))
+ assert 'foobar' not in t.dialect_options['participating2']
+
+ def test_contains_wildcard(self):
+ with self._fixture():
+ m = MetaData()
+ t = Table('x', m, Column('x', Integer), participating2_foobar=5)
+ assert 'foobar' in t.dialect_options['participating2']
+
+
+ def test_update(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c', participating_x=20)
+ eq_(idx.dialect_kwargs, {
+ "participating_x": 20,
+ 'participating_z_one': None,
+ "participating_y": False})
+ idx._validate_dialect_kwargs({
+ "participating_x": 25,
+ "participating_z_one": "default"})
+ eq_(idx.dialect_kwargs, {
+ "participating_x": 25,
+ 'participating_z_one': "default",
+ "participating_y": False})
+ idx._validate_dialect_kwargs({
+ "participating_x": 25,
+ "participating_z_one": "default"})
+ eq_(idx.dialect_kwargs, {
+ 'participating_z_one': 'default',
+ 'participating_y': False,
+ 'participating_x': 25})
+ idx._validate_dialect_kwargs({
+ "participating_y": True,
+ 'participating2_y': "p2y"})
+ eq_(idx.dialect_kwargs, {
+ "participating_x": 25,
+ "participating2_x": 9,
+ "participating_y": True,
+ 'participating2_y': "p2y",
+ "participating2_pp": "default",
+ "participating_z_one": "default"})
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
