return self
+
+class _DialectArgDictBase(object):
+ """base for dynamic dictionaries that handle dialect-level keyword
+ arguments."""
+
+ def _keys_iter(self):
+ raise NotImplementedError()
+ if util.py2k:
+ def keys(self):
+ return list(self._keys_iter())
+ def items(self):
+ return [(key, self[key]) for key in self._keys_iter()]
+ else:
+ def keys(self):
+ return self._keys_iter()
+ def items(self):
+ return ((key, self[key]) for key in self._keys_iter())
+
+ def get(self, key, default=None):
+ if key in self:
+ return self[key]
+ else:
+ return default
+
+ def __iter__(self):
+ return self._keys_iter()
+
+ def __eq__(self, other):
+ return dict(self) == dict(other)
+
+ def __repr__(self):
+ return repr(dict(self))
+
+class _DialectArgView(_DialectArgDictBase):
+ """A dictionary view of dialect-level arguments in the form
+ <dialectname>_<argument_name>.
+
+ """
+ def __init__(self, obj):
+ self.obj = obj
+
+ def __getitem__(self, key):
+ if "_" not in key:
+ raise KeyError(key)
+ dialect, value_key = key.split("_", 1)
+
+ try:
+ opt = self.obj.dialect_options[dialect]
+ except exc.NoSuchModuleError:
+ raise KeyError(key)
+ else:
+ return opt[value_key]
+
+ def __setitem__(self, key, value):
+ if "_" not in key:
+ raise exc.ArgumentError(
+ "Keys must be of the form <dialectname>_<argname>")
+
+ dialect, value_key = key.split("_", 1)
+ self.obj.dialect_options[dialect][value_key] = value
+
+ def _keys_iter(self):
+ return (
+ "%s_%s" % (dialect_name, value_name)
+ for dialect_name in self.obj.dialect_options
+ for value_name in self.obj.dialect_options[dialect_name]._non_defaults
+ )
+
+class _DialectArgDict(_DialectArgDictBase):
+ """A dictionary view of dialect-level arguments for a specific
+ dialect.
+
+ Maintains a separate collection of user-specified arguments
+ and dialect-specified default arguments.
+
+ """
+ def __init__(self, obj, dialect_name):
+ self._non_defaults = {}
+ self._defaults = {}
+
+ def _keys_iter(self):
+ return iter(set(self._non_defaults).union(self._defaults))
+
+ def __getitem__(self, key):
+ if key in self._non_defaults:
+ return self._non_defaults[key]
+ else:
+ return self._defaults[key]
+
+ def __setitem__(self, key, value):
+ self._non_defaults[key] = value
+
class DialectKWArgs(object):
"""Establish the ability for a class to have dialect-specific arguments
with defaults and validation.
"""
+ @classmethod
+ def argument_for(cls, dialect_name, argument_name, default):
+ """Add a new kind of dialect-specific keyword argument for this class.
+
+ E.g.::
+
+ Index.argument_for("mydialect", "length", None)
+
+ some_index = Index('a', 'b', mydialect_length=5)
+
+ The :meth:`.DialectKWArgs.argument_for` method is a per-argument
+ way adding extra arguments to the :attr:`.Dialect.construct_arguments`
+ dictionary. This dictionary provides a list of argument names accepted by
+ various schema-level constructs on behalf of a dialect.
+
+ New dialects should typically specify this dictionary all at once as a data
+ member of the dialect class. The use case for ad-hoc addition of
+ argument names is typically for end-user code that is also using
+ a custom compilation scheme which consumes the additional arguments.
+
+ :param dialect_name: name of a dialect. The dialect must be locatable,
+ else a :class:`.NoSuchModuleError` is raised. The dialect must
+ also include an existing :attr:`.Dialect.construct_arguments` collection,
+ indicating that it participates in the keyword-argument validation and
+ default system, else :class:`.ArgumentError` is raised.
+ If the dialect does not include this collection, then any keyword argument
+ can be specified on behalf of this dialect already. All dialects
+ packaged within SQLAlchemy include this collection, however for third
+ party dialects, support may vary.
+
+ :param argument_name: name of the parameter.
+
+ :param default: default value of the parameter.
+
+ .. versionadded:: 0.9.4
+
+ """
+
+ construct_arg_dictionary = DialectKWArgs._kw_registry[dialect_name]
+ if construct_arg_dictionary is None:
+ raise exc.ArgumentError("Dialect '%s' does have keyword-argument "
+ "validation and defaults enabled configured" %
+ dialect_name)
+ construct_arg_dictionary[cls][argument_name] = default
+
@util.memoized_property
def dialect_kwargs(self):
"""A collection of keyword arguments specified as dialect-specific
unlike the :attr:`.DialectKWArgs.dialect_options` collection, which
contains all options known by this dialect including defaults.
+ The collection is also writable; keys are accepted of the
+ form ``<dialect>_<kwarg>`` where the value will be assembled
+ into the list of options.
+
.. versionadded:: 0.9.2
+ .. versionchanged:: 0.9.4 The :attr:`.DialectKWArgs.dialect_kwargs`
+ collection is now writable.
+
.. seealso::
:attr:`.DialectKWArgs.dialect_options` - nested dictionary form
"""
-
- return util.immutabledict()
+ return _DialectArgView(self)
@property
def kwargs(self):
- """Deprecated; see :attr:`.DialectKWArgs.dialect_kwargs"""
+ """A synonym for :attr:`.DialectKWArgs.dialect_kwargs`."""
return self.dialect_kwargs
@util.dependencies("sqlalchemy.dialects")
def _kw_reg_for_dialect_cls(self, dialect_name):
construct_arg_dictionary = DialectKWArgs._kw_registry[dialect_name]
+ d = _DialectArgDict(self, dialect_name)
+
if construct_arg_dictionary is None:
- return {"*": None}
+ d._defaults.update({"*": None})
else:
- d = {}
for cls in reversed(self.__class__.__mro__):
if cls in construct_arg_dictionary:
- d.update(construct_arg_dictionary[cls])
- return d
+ d._defaults.update(construct_arg_dictionary[cls])
+ return d
@util.memoized_property
def dialect_options(self):
if not kwargs:
return
- self.dialect_kwargs = self.dialect_kwargs.union(kwargs)
-
for k in kwargs:
m = re.match('^(.+?)_(.+)$', k)
- if m is None:
+ if not m:
raise TypeError("Additional arguments should be "
"named <dialectname>_<argument>, got '%s'" % k)
dialect_name, arg_name = m.group(1, 2)
"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]}
+ self.dialect_options[dialect_name] = d = \
+ _DialectArgDict(self, dialect_name)
+ d._defaults.update({"*": None})
+ d._non_defaults[arg_name] = kwargs[k]
else:
if "*" not in construct_arg_dictionary and \
arg_name not in construct_arg_dictionary:
with mock.patch("sqlalchemy.dialects.registry.load", load):
yield
+ def teardown(self):
+ Index._kw_registry.clear()
+
def test_participating(self):
with self._fixture():
idx = Index('a', 'b', 'c', participating_y=True)
}
)
+ def test_bad_kwarg_raise(self):
+ with self._fixture():
+ assert_raises_message(
+ TypeError,
+ "Additional arguments should be named "
+ "<dialectname>_<argument>, got 'foobar'",
+ Index, 'a', 'b', 'c', foobar=True
+ )
def test_unknown_dialect_warning(self):
with self._fixture():
assert_raises_message(
'participating2_y': "p2y",
"participating_z_one": "default"})
+ def test_key_error_kwargs_no_dialect(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ assert_raises(
+ KeyError,
+ idx.kwargs.__getitem__, 'foo_bar'
+ )
+
+ def test_key_error_kwargs_no_underscore(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ assert_raises(
+ KeyError,
+ idx.kwargs.__getitem__, 'foobar'
+ )
+
+ def test_key_error_kwargs_no_argument(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ assert_raises(
+ KeyError,
+ idx.kwargs.__getitem__, 'participating_asdmfq34098'
+ )
+
+ assert_raises(
+ KeyError,
+ idx.kwargs.__getitem__, 'nonparticipating_asdmfq34098'
+ )
+
+ def test_key_error_dialect_options(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ assert_raises(
+ KeyError,
+ idx.dialect_options['participating'].__getitem__, 'asdfaso890'
+ )
+
+ assert_raises(
+ KeyError,
+ idx.dialect_options['nonparticipating'].__getitem__, 'asdfaso890'
+ )
+
+ def test_ad_hoc_participating_via_opt(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ idx.dialect_options['participating']['foobar'] = 5
+
+ eq_(idx.dialect_options['participating']['foobar'], 5)
+ eq_(idx.kwargs['participating_foobar'], 5)
+
+ def test_ad_hoc_nonparticipating_via_opt(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ idx.dialect_options['nonparticipating']['foobar'] = 5
+
+ eq_(idx.dialect_options['nonparticipating']['foobar'], 5)
+ eq_(idx.kwargs['nonparticipating_foobar'], 5)
+
+ def test_ad_hoc_participating_via_kwargs(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ idx.kwargs['participating_foobar'] = 5
+
+ eq_(idx.dialect_options['participating']['foobar'], 5)
+ eq_(idx.kwargs['participating_foobar'], 5)
+
+ def test_ad_hoc_nonparticipating_via_kwargs(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ idx.kwargs['nonparticipating_foobar'] = 5
+
+ eq_(idx.dialect_options['nonparticipating']['foobar'], 5)
+ eq_(idx.kwargs['nonparticipating_foobar'], 5)
+
+ def test_ad_hoc_via_kwargs_invalid_key(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ assert_raises_message(
+ exc.ArgumentError,
+ "Keys must be of the form <dialectname>_<argname>",
+ idx.kwargs.__setitem__, "foobar", 5
+ )
+
+ def test_ad_hoc_via_kwargs_invalid_dialect(self):
+ with self._fixture():
+ idx = Index('a', 'b', 'c')
+ assert_raises_message(
+ exc.ArgumentError,
+ "no dialect 'nonexistent'",
+ idx.kwargs.__setitem__, "nonexistent_foobar", 5
+ )
+
+ def test_add_new_arguments_participating(self):
+ with self._fixture():
+ Index.argument_for("participating", "xyzqpr", False)
+
+ idx = Index('a', 'b', 'c', participating_xyzqpr=True)
+
+ eq_(idx.kwargs['participating_xyzqpr'], True)
+
+ idx = Index('a', 'b', 'c')
+ eq_(idx.dialect_options['participating']['xyzqpr'], False)
+
+ def test_add_new_arguments_nonparticipating(self):
+ with self._fixture():
+ assert_raises_message(
+ exc.ArgumentError,
+ "Dialect 'nonparticipating' does have keyword-argument "
+ "validation and defaults enabled configured",
+ Index.argument_for, "nonparticipating", "xyzqpr", False
+ )
+
+
+ def test_add_new_arguments_invalid_dialect(self):
+ with self._fixture():
+ assert_raises_message(
+ exc.ArgumentError,
+ "no dialect 'nonexistent'",
+ Index.argument_for, "nonexistent", "foobar", 5
+ )
+
+
class NamingConventionTest(fixtures.TestBase):
def _fixture(self, naming_convention, table_schema=None):
m1 = MetaData(naming_convention=naming_convention)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
