.. changelog::
:version: 0.9.0b2
+ .. change::
+ :tags: feature, engine
+ :tickets: 2875
+
+ The :func:`.engine_from_config` function has been improved so that
+ we will be able to parse dialect-specific arguments from string
+ configuration dictionaries. Dialect classes can now provide their
+ own list of parameter types and string-conversion routines.
+ The feature is not yet used by the built-in dialects, however.
+
.. change::
:tags: bug, sql
:tickets: 2879
arguments.
"""
- opts = util._coerce_config(configuration, prefix)
- opts.update(kwargs)
- url = opts.pop('url')
- return create_engine(url, **opts)
+ options = dict((key[len(prefix):], configuration[key])
+ for key in configuration
+ if key.startswith(prefix))
+ options['_coerce_config'] = True
+ options.update(kwargs)
+ url = options.pop('url')
+ return create_engine(url, **options)
__all__ = (
supports_simple_order_by_label = True
+ engine_config_types = util.immutabledict([
+ ('convert_unicode', util.bool_or_str('force')),
+ ('pool_timeout', int),
+ ('echo', util.bool_or_str('debug')),
+ ('echo_pool', util.bool_or_str('debug')),
+ ('pool_recycle', int),
+ ('pool_size', int),
+ ('max_overflow', int),
+ ('pool_threadlocal', bool),
+ ('use_native_unicode', bool),
+ ])
+
# if the NUMERIC type
# returns decimal.Decimal.
# *not* the FLOAT type however.
dialect_cls = u.get_dialect()
+ if kwargs.pop('_coerce_config', False):
+ def pop_kwarg(key, default=None):
+ value = kwargs.pop(key, default)
+ if key in dialect_cls.engine_config_types:
+ value = dialect_cls.engine_config_types[key](value)
+ return value
+ else:
+ pop_kwarg = kwargs.pop
+
dialect_args = {}
# consume dialect arguments from kwargs
for k in util.get_cls_kwargs(dialect_cls):
if k in kwargs:
- dialect_args[k] = kwargs.pop(k)
+ dialect_args[k] = pop_kwarg(k)
dbapi = kwargs.pop('module', None)
if dbapi is None:
dbapi_args = {}
for k in util.get_func_kwargs(dialect_cls.dbapi):
if k in kwargs:
- dbapi_args[k] = kwargs.pop(k)
+ dbapi_args[k] = pop_kwarg(k)
dbapi = dialect_cls.dbapi(**dbapi_args)
dialect_args['dbapi'] = dbapi
# assemble connection arguments
(cargs, cparams) = dialect.create_connect_args(u)
- cparams.update(kwargs.pop('connect_args', {}))
+ cparams.update(pop_kwarg('connect_args', {}))
# look for existing pool or create
- pool = kwargs.pop('pool', None)
+ pool = pop_kwarg('pool', None)
if pool is None:
def connect():
try:
)
)
- creator = kwargs.pop('creator', connect)
+ creator = pop_kwarg('creator', connect)
- poolclass = kwargs.pop('poolclass', None)
+ poolclass = pop_kwarg('poolclass', None)
if poolclass is None:
poolclass = dialect_cls.get_pool_class(u)
pool_args = {}
for k in util.get_cls_kwargs(poolclass):
tk = translate.get(k, k)
if tk in kwargs:
- pool_args[k] = kwargs.pop(tk)
+ pool_args[k] = pop_kwarg(tk)
pool = poolclass(creator, **pool_args)
else:
if isinstance(pool, poollib._DBProxy):
engine_args = {}
for k in util.get_cls_kwargs(engineclass):
if k in kwargs:
- engine_args[k] = kwargs.pop(k)
+ engine_args[k] = pop_kwarg(k)
_initialize = kwargs.pop('_initialize', True)
from .. import util
-
-def _coerce_config(configuration, prefix):
- """Convert configuration values to expected types."""
-
- options = dict((key[len(prefix):], configuration[key])
- for key in configuration
- if key.startswith(prefix))
- for option, type_ in (
- ('convert_unicode', util.bool_or_str('force')),
- ('pool_timeout', int),
- ('echo', util.bool_or_str('debug')),
- ('echo_pool', util.bool_or_str('debug')),
- ('pool_recycle', int),
- ('pool_size', int),
- ('max_overflow', int),
- ('pool_threadlocal', bool),
- ('use_native_unicode', bool),
- ):
- util.coerce_kw_type(options, option, type_)
- return options
-
-
def connection_memoize(key):
"""Decorator, memoize a function in a connection.info stash.
from sqlalchemy.util.compat import configparser, StringIO
import sqlalchemy.engine.url as url
from sqlalchemy import create_engine, engine_from_config, exc, pool
-from sqlalchemy.engine.util import _coerce_config
from sqlalchemy.engine.default import DefaultDialect
import sqlalchemy as tsa
from sqlalchemy.testing import fixtures
from sqlalchemy import testing
-from sqlalchemy.testing.mock import Mock
+from sqlalchemy.testing.mock import Mock, MagicMock, patch
class ParseConnectTest(fixtures.TestBase):
module=dbapi, _initialize=False)
c = e.connect()
- def test_coerce_config(self):
- raw = r"""
-[prefixed]
-sqlalchemy.url=postgresql://scott:tiger@somehost/test?fooz=somevalue
-sqlalchemy.convert_unicode=0
-sqlalchemy.echo=false
-sqlalchemy.echo_pool=1
-sqlalchemy.max_overflow=2
-sqlalchemy.pool_recycle=50
-sqlalchemy.pool_size=2
-sqlalchemy.pool_threadlocal=1
-sqlalchemy.pool_timeout=10
-[plain]
-url=postgresql://scott:tiger@somehost/test?fooz=somevalue
-convert_unicode=0
-echo=0
-echo_pool=1
-max_overflow=2
-pool_recycle=50
-pool_size=2
-pool_threadlocal=1
-pool_timeout=10
-"""
- ini = configparser.ConfigParser()
- ini.readfp(StringIO(raw))
-
- expected = {
- 'url': 'postgresql://scott:tiger@somehost/test?fooz=somevalue',
- 'convert_unicode': 0,
- 'echo': False,
- 'echo_pool': True,
- 'max_overflow': 2,
- 'pool_recycle': 50,
- 'pool_size': 2,
- 'pool_threadlocal': True,
- 'pool_timeout': 10,
- }
-
- prefixed = dict(ini.items('prefixed'))
- self.assert_(_coerce_config(prefixed, 'sqlalchemy.')
- == expected)
-
- plain = dict(ini.items('plain'))
- self.assert_(_coerce_config(plain, '') == expected)
def test_engine_from_config(self):
dbapi = mock_dbapi
'z=somevalue')
assert e.echo is True
- for param, values in [
- ('convert_unicode', ('true', 'false', 'force')),
- ('echo', ('true', 'false', 'debug')),
- ('echo_pool', ('true', 'false', 'debug')),
- ('use_native_unicode', ('true', 'false')),
- ]:
- for value in values:
- config = {
- 'sqlalchemy.url': 'postgresql://scott:tiger@somehost/test',
- 'sqlalchemy.%s' % param : value
- }
- cfg = _coerce_config(config, 'sqlalchemy.')
- assert cfg[param] == {'true':True, 'false':False}.get(value, value)
+
+ def test_engine_from_config_custom(self):
+ from sqlalchemy import util
+ from sqlalchemy.dialects import registry
+ tokens = __name__.split(".")
+
+ class MyDialect(MockDialect):
+ engine_config_types = {
+ "foobar": int,
+ "bathoho": util.bool_or_str('force')
+ }
+
+ def __init__(self, foobar=None, bathoho=None, **kw):
+ self.foobar = foobar
+ self.bathoho = bathoho
+
+ global dialect
+ dialect = MyDialect
+ registry.register("mockdialect.barb",
+ ".".join(tokens[0:-1]), tokens[-1])
+
+ config = {
+ "sqlalchemy.url": "mockdialect+barb://",
+ "sqlalchemy.foobar": "5",
+ "sqlalchemy.bathoho": "false"
+ }
+ e = engine_from_config(config, _initialize=False)
+ eq_(e.dialect.foobar, 5)
+ eq_(e.dialect.bathoho, False)
def test_custom(self):
)
return connection
- return Mock(
+ return MagicMock(
sqlite_version_info=(99, 9, 9,),
version_info=(99, 9, 9,),
sqlite_version='99.9.9',
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
