.. changelog::
:version: 1.0.5
+ .. change::
+ :tags: bug, engine
+ :tickets: 3435
+
+ Fixed bug where known boolean values used by
+ :func:`.engine_from_config` were not being parsed correctly;
+ these included ``pool_threadlocal`` and the psycopg2 argument
+ ``use_native_unicode``.
+
.. change::
:tags: bug, mssql
:tickets: 3424, 3430
sqltypes.Time: TIME,
}
+ engine_config_types = default.DefaultDialect.engine_config_types.union([
+ ('legacy_schema_aliasing', util.asbool),
+ ])
+
ischema_names = ischema_names
supports_native_boolean = False
_has_native_json = False
_has_native_jsonb = False
+ engine_config_types = PGDialect.engine_config_types.union([
+ ('use_native_unicode', util.asbool),
+ ])
+
colspecs = util.update_copy(
PGDialect.colspecs,
{
'prefix' argument indicates the prefix to be searched for.
A select set of keyword arguments will be "coerced" to their
- expected type based on string values. In a future release, this
- functionality will be expanded and include dialect-specific
- arguments.
+ expected type based on string values. The set of arguments
+ is extensible per-dialect using the ``engine_config_types`` accessor.
+
"""
options = dict((key[len(prefix):], configuration[key])
engine_config_types = util.immutabledict([
('convert_unicode', util.bool_or_str('force')),
- ('pool_timeout', int),
+ ('pool_timeout', util.asint),
('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),
+ ('pool_recycle', util.asint),
+ ('pool_size', util.asint),
+ ('max_overflow', util.asint),
+ ('pool_threadlocal', util.asbool),
])
# if the NUMERIC type
engine.connect)
+class EngineFromConfigTest(fixtures.TestBase):
+ def test_legacy_schema_flag(self):
+ cfg = {
+ "sqlalchemy.url": "mssql://foodsn",
+ "sqlalchemy.legacy_schema_aliasing": "false"
+ }
+ e = engine_from_config(cfg)
+ eq_(e.dialect.legacy_schema_aliasing, False)
+
+
class VersionDetectionTest(fixtures.TestBase):
def test_pymssql_version(self):
dialect = pymssql.MSDialect_pymssql()
import logging
import logging.handlers
from sqlalchemy.testing.mock import Mock
+from sqlalchemy.engine import engine_from_config
class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
new_encoding = c.execute("show client_encoding").fetchone()[0]
eq_(new_encoding, test_encoding)
+ def test_pg_dialect_use_native_unicode_from_config(self):
+ config = {
+ 'sqlalchemy.url': 'postgresql://scott:tiger@somehost/test',
+ 'sqlalchemy.use_native_unicode': "false"}
+
+ e = engine_from_config(config, _initialize=False)
+ eq_(e.dialect.use_native_unicode, False)
+
+ config = {
+ 'sqlalchemy.url': 'postgresql://scott:tiger@somehost/test',
+ 'sqlalchemy.use_native_unicode': "true"}
+
+ e = engine_from_config(config, _initialize=False)
+ eq_(e.dialect.use_native_unicode, True)
+
+
@testing.only_on(
['postgresql+psycopg2', 'postgresql+pg8000',
'postgresql+psycopg2cffi'],
'z=somevalue')
assert e.echo is True
+ def test_pool_threadlocal_from_config(self):
+ dbapi = mock_dbapi
+
+ config = {
+ 'sqlalchemy.url': 'postgresql://scott:tiger@somehost/test',
+ 'sqlalchemy.pool_threadlocal': "false"}
+
+ e = engine_from_config(config, module=dbapi, _initialize=False)
+ eq_(e.pool._use_threadlocal, False)
+
+ config = {
+ 'sqlalchemy.url': 'postgresql://scott:tiger@somehost/test',
+ 'sqlalchemy.pool_threadlocal': "true"}
+
+ e = engine_from_config(config, module=dbapi, _initialize=False)
+ eq_(e.pool._use_threadlocal, True)
+
def test_pool_reset_on_return_from_config(self):
dbapi = mock_dbapi