Commits

Mike Bayer committed 53bb115

added engine_from_config() function for helping to create_engine()
from an .ini style config

Comments (0)

Files changed (4)

lib/sqlalchemy/__init__.py

 from sqlalchemy.types import *
 from sqlalchemy.sql import *
 from sqlalchemy.schema import *
-from sqlalchemy.engine import create_engine
+from sqlalchemy.engine import create_engine, engine_from_config
 
 
 __version__ = 'svn'

lib/sqlalchemy/engine/__init__.py

 import sqlalchemy.databases
 from sqlalchemy.engine.base import *
 from sqlalchemy.engine import strategies
+from sqlalchemy import util
 
 def engine_descriptors():
     """Provide a listing of all the database implementations supported.
     strategy = kwargs.pop('strategy', default_strategy)
     strategy = strategies.strategies[strategy]
     return strategy.create(*args, **kwargs)
+
+def engine_from_config(configuration, prefix='sqlalchemy.', **kwargs):
+    """Create a new Engine instance using a configuration dictionary.
+    
+    the dictionary is typically produced from a config file where keys are prefixed,
+    such as sqlalchemy.url, sqlalchemy.echo, etc.  The '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 to include 
+    dialect-specific arguments.
+    """
+    
+    opts = dict([(key[len(prefix):], configuration[key]) for key in configuration if key.startswith(prefix)])
+    for opt, type_ in (
+        ('convert_unicode', bool),
+        ('pool_timeout', int),
+        ('echo', bool),
+        ('echo_pool', bool),
+        ('pool_recycle', int),
+    ):
+        util.coerce_kw_type(opts, opt, type_)
+    opts.update(kwargs)
+    url = opts.pop('url')
+    return create_engine(url, **opts)
+
+    

lib/sqlalchemy/engine/url.py

             keys.sort()
             s += '?' + "&".join(["%s=%s" % (k, self.query[k]) for k in keys])
         return s
-
+    
+    def __eq__(self, other):
+        return \
+            isinstance(other, URL) and \
+            self.drivername == other.drivername and \
+            self.username == other.username and \
+            self.password == other.password and \
+            self.host == other.host and \
+            self.database == other.database and \
+            self.query == other.query
+            
     def get_dialect(self):
         """Return the SQLAlchemy database dialect class corresponding to this URL's driver name."""
         dialect=None

test/engine/parseconnect.py

 
         
 class ParseConnectTest(PersistTest):
-    def testrfc1738(self):
+    def test_rfc1738(self):
         for text in (
             'dbtype://username:password@hostspec:110//usr/db_file.db',
             'dbtype://username:password@hostspec/database',
 
 class CreateEngineTest(PersistTest):
     """test that create_engine arguments of different types get propigated properly"""
-    def testconnectquery(self):
+    def test_connect_query(self):
         dbapi = MockDBAPI(foober='12', lala='18', fooz='somevalue')
         
         # start the postgres dialect, but put our mock DBAPI as the module instead of psycopg
         e = create_engine('postgres://scott:tiger@somehost/test?foober=12&lala=18&fooz=somevalue', module=dbapi)
         c = e.connect()
 
-    def testkwargs(self):
+    def test_kwargs(self):
         dbapi = MockDBAPI(foober=12, lala=18, hoho={'this':'dict'}, fooz='somevalue')
 
         # start the postgres dialect, but put our mock DBAPI as the module instead of psycopg
         e = create_engine('postgres://scott:tiger@somehost/test?fooz=somevalue', connect_args={'foober':12, 'lala':18, 'hoho':{'this':'dict'}}, module=dbapi)
         c = e.connect()
 
-    def testcustom(self):
+    def test_engine_from_config(self):
+        dbapi = MockDBAPI()
+
+        config = {
+            'sqlalchemy.url':'postgres://scott:tiger@somehost/test?fooz=somevalue',
+            'sqlalchemy.echo':'1',
+            'sqlalchemy.pool_recycle':50
+        }
+
+        e = engine_from_config(config, module=dbapi)
+        assert e.pool._recycle == 50
+        assert e.echo is True
+        assert e.url == url.make_url('postgres://scott:tiger@somehost/test?fooz=somevalue')
+        
+    def test_custom(self):
         dbapi = MockDBAPI(foober=12, lala=18, hoho={'this':'dict'}, fooz='somevalue')
 
         def connect():
         e = create_engine('postgres://', creator=connect, module=dbapi)
         c = e.connect()
     
-    def testrecycle(self):
+    def test_recycle(self):
         dbapi = MockDBAPI(foober=12, lala=18, hoho={'this':'dict'}, fooz='somevalue')
         e = create_engine('postgres://', pool_recycle=472, module=dbapi)
         assert e.pool._recycle == 472
         
-    def testbadargs(self):
+    def test_badargs(self):
         # good arg, use MockDBAPI to prevent oracle import errors
         e = create_engine('oracle://', use_ansi=True, module=MockDBAPI())
         
         except exceptions.DBAPIError:
             assert True
     
-    def testurlattr(self):
+    def test_urlattr(self):
         """test the url attribute on ``Engine``."""
         
         e = create_engine('mysql://scott:tiger@localhost/test', module=MockDBAPI())
         assert e.url.username == e2.url.username == 'scott'
         assert e2.url is u
         
-    def testpoolargs(self):
+    def test_poolargs(self):
         """test that connection pool args make it thru"""
         e = create_engine('postgres://', creator=None, pool_recycle=50, echo_pool=None, module=MockDBAPI())
         assert e.pool._recycle == 50