rattail-project/rattail Changeset

Changeset - 87c70c437565

Parent rev.

Child rev.

[Not reviewed]

0 3 1

Lance Edgar (lance) - 10 years ago 2014-05-31 00:16:33
lance@edbob.org

Accept config section name within `db.util.get_engines()`.

This is to set the stage for leveraging this function to establish database
connections to other systems, e.g. for `rattail_locsms`.

4 files changed with 114 insertions and 45 deletions:

docs/api/rattail/db/util.rst

docs/index.rst

rattail/db/util.py

tests/db/test_util.py

0 comments (0 inline, 0 general)

docs/api/rattail/db/util.rst

➞

Show inline comments

@@ new file 100644 @@
 .. -*- coding: utf-8 -*-
 ``rattail.db.util``
 ===================
 .. automodule:: rattail.db.util
    :members:

docs/index.rst

➞

Show inline comments

@@ @@ -13,12 +13,13 @@ some attention thus far. @@
 Package API:
 .. toctree::
    :maxdepth: 1
    api/rattail
    api/rattail/db/util
    api/rattail/enum
    api/rattail/exceptions
    api/rattail/filemon/util
    api/rattail/logging

rattail/db/util.py

➞

Show inline comments

 #!/usr/bin/env python
-# -*- coding: utf-8  -*-
 # -*- coding: utf-8 -*-
 ################################################################################
+#
 #  Rattail -- Retail Software Framework
-#  Copyright © 2010-2012 Lance Edgar
+#  Copyright © 2010-2014 Lance Edgar
+#
 #  This file is part of Rattail.
+#
 #  Rattail is free software: you can redistribute it and/or modify it under the
 #  terms of the GNU Affero General Public License as published by the Free
 #  Software Foundation, either version 3 of the License, or (at your option)
@@ @@ -23,12 +23,14 @@ @@
 ################################################################################
 """
 Database Utilities
 """
 import warnings
 from sqlalchemy import engine_from_config as sa_engine_from_config
 from ..util import load_object
 def engine_from_config(configuration, prefix='sqlalchemy.', **kwargs):
     if key in config:
         kwargs.setdefault('poolclass', load_object(config[key]))
         del config[key]
     return sa_engine_from_config(config, prefix, **kwargs)
 def get_engines(config):
+def get_engines(config, section=None):
     """
     Fetch all database engines defined in the given config object.
     Fetch all database engines defined in the given config object for a given
     section.
     :param config: A ``ConfigParser`` instance containing app configuration.
     :type section: string
     :param section: Optional section name within which the configuration
        options are defined.  If not specified, ``'rattail.db'`` is assumed.
     :returns: A dictionary of SQLAlchemy engine instances, keyed according to
        the config settings.
     .. note::
        Until the ``edbob`` dependency is fully removed, ``'edbob.db'`` will be
        considered as a fallback section if none is specified and there is no
        ``'rattail.db'`` section (or it does not contain any configuration).  If
        this fallback is utilized, a deprecation warning will be emitted.
     """
     keys = config.get('edbob.db', 'keys')
     if keys:
         keys = keys.split(',')
     else:
         keys = ['default']
     engines = {}
     cfg = config.get_dict('edbob.db')
     for key in keys:
         key = key.strip()
         try:
             engines[key] = engine_from_config(cfg, prefix='{0}.'.format(key))
         except KeyError:
             if key == 'default':
                 try:
                     engines[key] = engine_from_config(cfg, prefix='sqlalchemy.')
                 except KeyError:
                     pass
     def engines_from_section(section):
         keys = config.get(section, u'keys')
         if keys:
             keys = keys.split(u',')
         else:
             keys = [u'default']
         engines = {}
         cfg = config.get_dict(section)
         for key in keys:
             key = key.strip()
             try:
                 engines[key] = engine_from_config(cfg, prefix=u'{0}.'.format(key))
             except KeyError:
                 if key == u'default':
                     try:
                         engines[key] = engine_from_config(cfg, prefix=u'sqlalchemy.')
                     except KeyError:
                         pass
         return engines
     if section is not None:
         return engines_from_section(section)
     engines = engines_from_section(u'rattail.db')
     if engines:
         return engines
     engines = engines_from_section(u'edbob.db')
     if engines:
         warnings.warn(u"Defining database engines in [edbob.db] is deprecated; please "
                       u"use the [rattail.db] section instead.", DeprecationWarning)
     return engines

tests/db/test_util.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import warnings
 from unittest import TestCase
 from sqlalchemy.pool import SingletonThreadPool, NullPool
 from edbob.configuration import AppConfigParser
@@ @@ -24,31 +26,62 @@ class TestEngineConfig(TestCase): @@
             'sqlalchemy.poolclass': 'sqlalchemy.pool:NullPool',
+            }
         engine = util.engine_from_config(config)
         self.assertEqual(str(engine.url), 'sqlite://')
         self.assertTrue(isinstance(engine.pool, NullPool))
     def test_get_engines_default(self):
         config = AppConfigParser('rattail')
         config.set('edbob.db', 'sqlalchemy.url', 'sqlite://')
         engines = util.get_engines(config)
 class TestGetEngines(TestCase):
     def setUp(self):
         self.config = AppConfigParser(u'rattail')
     def test_default_section_is_rattail_db(self):
         self.config.set(u'rattail.db', u'keys', u'default')
         self.config.set(u'rattail.db', u'default.url', u'sqlite://')
         engines = util.get_engines(self.config)
         self.assertEqual(len(engines), 1)
         self.assertEqual(engines.keys()[0], u'default')
         self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
     def test_custom_section_is_honored(self):
         self.config.set(u'mycustomdb', u'keys', u'default')
         self.config.set(u'mycustomdb', u'default.url', u'sqlite://')
         engines = util.get_engines(self.config, section=u'mycustomdb')
         self.assertEqual(len(engines), 1)
         self.assertEqual(engines.keys()[0], u'default')
         self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
     def test_default_section_falls_back_to_edbob_db(self):
         self.config.set(u'edbob.db', u'keys', u'default')
         self.config.set(u'edbob.db', u'default.url', u'sqlite://')
         with warnings.catch_warnings(record=True) as ignore_warnings:
             engines = util.get_engines(self.config)
         self.assertEqual(len(engines), 1)
         self.assertEqual(engines.keys()[0], u'default')
         self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
     def test_default_prefix_does_not_require_keys_declaration(self):
         self.config.set(u'rattail.db', u'default.url', u'sqlite://')
         engines = util.get_engines(self.config)
         self.assertEqual(len(engines), 1)
         self.assertEqual(str(engines['default'].url), 'sqlite://')
     def test_get_engines_custom(self):
         config = AppConfigParser('rattail')
         config.set('edbob.db', 'keys', 'host, store')
         config.set('edbob.db', 'host.url', 'sqlite:///rattail.host.sqlite')
         config.set('edbob.db', 'store.url', 'sqlite:///rattail.store.sqlite')
         config.set('edbob.db', 'store.poolclass', 'sqlalchemy.pool:SingletonThreadPool')
         engines = util.get_engines(config)
         self.assertEqual(engines.keys()[0], u'default')
         self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
     def test_default_prefix_falls_back_to_sqlalchemy(self):
         # Still no need to define "keys" option here.
         self.config.set(u'rattail.db', u'sqlalchemy.url', u'sqlite://')
         engines = util.get_engines(self.config)
         self.assertEqual(len(engines), 1)
         self.assertEqual(engines.keys()[0], u'default')
         self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
     def test_defined_keys_are_included_in_engines_result(self):
         # Note there is no "default" key here.
         self.config.set(u'rattail.db', u'keys', u'host, store')
         self.config.set(u'rattail.db', u'host.url', u'sqlite:///rattail.host.sqlite')
         self.config.set(u'rattail.db', u'store.url', u'sqlite:///rattail.store.sqlite')
         engines = util.get_engines(self.config)
         self.assertEqual(len(engines), 2)
         self.assertEqual(str(engines['host'].url), 'sqlite:///rattail.host.sqlite')
         self.assertTrue(isinstance(engines['host'].pool, NullPool))
         self.assertEqual(str(engines['store'].url), 'sqlite:///rattail.store.sqlite')
         self.assertTrue(isinstance(engines['store'].pool, SingletonThreadPool))
     def test_get_engines_none(self):
         config = AppConfigParser('rattail')
         config.set('edbob.db', 'unknown.url', 'sqlite://')
         engines = util.get_engines(config)
         self.assertEqual(len(engines), 0)
         self.assertEqual(sorted(engines.keys()), [u'host', u'store'])
         self.assertEqual(unicode(engines[u'host'].url), u'sqlite:///rattail.host.sqlite')
         self.assertEqual(unicode(engines[u'store'].url), u'sqlite:///rattail.store.sqlite')

0 comments (0 inline, 0 general)