github/sprockets.clients.postgresql
1
0
Fork 0
mirror of https://github.com/sprockets/sprockets.clients.postgresql.git synced 2024-11-22 03:00:21 +00:00
Code Issues Projects Releases Packages Wiki Activity

Add db_url parameter to session classes.

Browse source
This commit is contained in:
Dave Shawley 2014-11-14 08:11:07 -05:00
parent 2e9749fae2
commit 336c47dd9a
2 changed files with 36 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
sprockets/clients/postgresql/__init__.py
View file

@ -95,13 +95,18 @@ class Session(queries.Session):
:param queries.cursor: The cursor type to use
:param int pool_idle_ttl: How long idle pools keep connections open
:param int pool_max_size: The maximum size of the pool to use
:param str db_url: Optional database connection URL. Use this when
you need to connect to a database that is only known at runtime.
"""
def __init__(self, dbname,
cursor_factory=queries.RealDictCursor,
pool_idle_ttl=pool.DEFAULT_IDLE_TTL,
pool_max_size=pool.DEFAULT_MAX_SIZE):
super(Session, self).__init__(_get_uri(dbname),
pool_max_size=pool.DEFAULT_MAX_SIZE,
db_url=None):
if db_url is None:
db_url = _get_uri(dbname)
super(Session, self).__init__(db_url,
cursor_factory,
pool_idle_ttl,
pool_max_size)
@ -125,14 +130,18 @@ class TornadoSession(tornado_session.TornadoSession):
:param int pool_max_size: The maximum size of the pool to use
:param tornado.ioloop.IOLoop ioloop: Pass in the instance of the tornado
IOLoop you would like to use. Defaults to the global instance.
:param str db_url: Optional database connection URL. Use this when
you need to connect to a database that is only known at runtime.
"""
def __init__(self, dbname,
cursor_factory=queries.RealDictCursor,
pool_idle_ttl=pool.DEFAULT_IDLE_TTL,
pool_max_size=tornado_session.DEFAULT_MAX_POOL_SIZE,
io_loop=None):
super(TornadoSession, self).__init__(_get_uri(dbname),
io_loop=None, db_url=None):
if db_url is None:
db_url = _get_uri(dbname)
super(TornadoSession, self).__init__(db_url,
cursor_factory,
pool_idle_ttl,
pool_max_size,

32
tests.py
View file

@ -11,7 +11,8 @@ except ImportError:
import unittest
from sprockets.clients import postgresql
import queries
import queries.pool
import queries.tornado_session
from tornado import testing
@ -28,25 +29,38 @@ class TestGetURI(unittest.TestCase):
class TestSession(unittest.TestCase):
@mock.patch('queries.session.Session.__init__')
def setUp(self, mock_init):
self.mock_init = mock_init
def test_session_invokes_queries_session(self, mock_init):
os.environ['PGSQL_TEST2'] = 'postgresql://foo:baz@db1:5433/bar'
self.session = postgresql.Session('test2')
self.assertTrue(mock_init.called)
def test_session_invokes_queries_session(self):
self.assertTrue(self.mock_init.called)
@mock.patch('queries.session.Session.__init__')
def test_session_invokes_queries_session_with_url(self, mock_init):
self.session = postgresql.Session('test6',
db_url='postgresql://localhost/b')
mock_init.assert_called_once_with(
'postgresql://localhost/b', queries.RealDictCursor,
queries.pool.DEFAULT_IDLE_TTL, queries.pool.DEFAULT_MAX_SIZE,
)
class TestTornadoSession(unittest.TestCase):
@mock.patch('queries.tornado_session.TornadoSession.__init__')
def setUp(self, mock_init):
self.mock_init = mock_init
def test_session_invokes_queries_session(self, mock_init):
os.environ['PGSQL_TEST3'] = 'postgresql://foo:baz@db1:5434/bar'
self.session = postgresql.TornadoSession('test3')
self.assertTrue(mock_init.called)
def test_session_invokes_queries_session(self):
self.assertTrue(self.mock_init.called)
@mock.patch('queries.tornado_session.TornadoSession.__init__')
def test_session_invokes_queries_session_with_url(self, mock_init):
self.session = postgresql.TornadoSession(
'test7', db_url='postgresql://localhost/b')
mock_init.assert_called_once_with(
'postgresql://localhost/b', queries.RealDictCursor,
queries.pool.DEFAULT_IDLE_TTL,
queries.tornado_session.DEFAULT_MAX_POOL_SIZE, None,
)
class SessionIntegrationTests(unittest.TestCase):