"""PooledDB - pooling for DB-API 2 connections.
|
|
Implements a pool of steady, thread-safe cached connections
|
to a database which are transparently reused,
|
using an arbitrary DB-API 2 compliant database interface module.
|
|
This should result in a speedup for persistent applications such as the
|
application server of "Webware for Python," without loss of robustness.
|
|
Robustness is provided by using "hardened" SteadyDB connections.
|
Even if the underlying database is restarted and all connections
|
are lost, they will be automatically and transparently reopened.
|
However, since you don't want this to happen in the middle of a database
|
transaction, you must explicitly start transactions with the begin()
|
method so that SteadyDB knows that the underlying connection shall not
|
be replaced and errors passed on until the transaction is completed.
|
|
Measures are taken to make the pool of connections thread-safe.
|
If the underlying DB-API module is thread-safe at the connection level,
|
the requested connections may be shared with other threads by default,
|
but you can also request dedicated connections in case you need them.
|
|
For the Python DB-API 2 specification, see:
|
https://www.python.org/dev/peps/pep-0249/
|
For information on Webware for Python, see:
|
https://webwareforpython.github.io/w4py/
|
|
|
Usage:
|
|
First you need to set up the database connection pool by creating
|
an instance of PooledDB, passing the following parameters:
|
|
creator: either an arbitrary function returning new DB-API 2
|
connection objects or a DB-API 2 compliant database module
|
mincached: the initial number of idle connections in the pool
|
(the default of 0 means no connections are made at startup)
|
maxcached: the maximum number of idle connections in the pool
|
(the default value of 0 or None means unlimited pool size)
|
maxshared: maximum number of shared connections allowed
|
(the default value of 0 or None means all connections are dedicated)
|
When this maximum number is reached, connections are
|
shared if they have been requested as shareable.
|
maxconnections: maximum number of connections generally allowed
|
(the default value of 0 or None means any number of connections)
|
blocking: determines behavior when exceeding the maximum
|
(if this is set to true, block and wait until the number of
|
connections decreases, but by default an error will be reported)
|
maxusage: maximum number of reuses of a single connection
|
(the default of 0 or None means unlimited reuse)
|
When this maximum usage number of the connection is reached,
|
the connection is automatically reset (closed and reopened).
|
setsession: an optional list of SQL commands that may serve to
|
prepare the session, e.g. ["set datestyle to german", ...]
|
reset: how connections should be reset when returned to the pool
|
(False or None to rollback transcations started with begin(),
|
the default value True always issues a rollback for safety's sake)
|
failures: an optional exception class or a tuple of exception classes
|
for which the connection failover mechanism shall be applied,
|
if the default (OperationalError, InternalError) is not adequate
|
ping: an optional flag controlling when connections are checked
|
with the ping() method if such a method is available
|
(0 = None = never, 1 = default = whenever fetched from the pool,
|
2 = when a cursor is created, 4 = when a query is executed,
|
7 = always, and all other bit combinations of these values)
|
|
The creator function or the connect function of the DB-API 2 compliant
|
database module specified as the creator will receive any additional
|
parameters such as the host, database, user, password etc. You may
|
choose some or all of these parameters in your own creator function,
|
allowing for sophisticated failover and load-balancing mechanisms.
|
|
For instance, if you are using pgdb as your DB-API 2 database module and
|
want a pool of at least five connections to your local database 'mydb':
|
|
import pgdb # import used DB-API 2 module
|
from dbutils.pooled_db import PooledDB
|
pool = PooledDB(pgdb, 5, database='mydb')
|
|
Once you have set up the connection pool you can request
|
database connections from that pool:
|
|
db = pool.connection()
|
|
You can use these connections just as if they were ordinary
|
DB-API 2 connections. Actually what you get is the hardened
|
SteadyDB version of the underlying DB-API 2 connection.
|
|
Please note that the connection may be shared with other threads
|
by default if you set a non-zero maxshared parameter and the DB-API 2
|
module allows this. If you want to have a dedicated connection, use:
|
|
db = pool.connection(shareable=False)
|
|
You can also use this to get a dedicated connection:
|
|
db = pool.dedicated_connection()
|
|
If you don't need it any more, you should immediately return it to the
|
pool with db.close(). You can get another connection in the same way.
|
|
Warning: In a threaded environment, never do the following:
|
|
pool.connection().cursor().execute(...)
|
|
This would release the connection too early for reuse which may be
|
fatal if the connections are not thread-safe. Make sure that the
|
connection object stays alive as long as you are using it, like that:
|
|
db = pool.connection()
|
cur = db.cursor()
|
cur.execute(...)
|
res = cur.fetchone()
|
cur.close() # or del cur
|
db.close() # or del db
|
|
Note that you need to explicitly start transactions by calling the
|
begin() method. This ensures that the connection will not be shared
|
with other threads, that the transparent reopening will be suspended
|
until the end of the transaction, and that the connection will be rolled
|
back before being given back to the connection pool.
|
|
|
Ideas for improvement:
|
|
* Add a thread for monitoring, restarting (or closing) bad or expired
|
connections (similar to DBConnectionPool/ResourcePool by Warren Smith).
|
* Optionally log usage, bad connections and exceeding of limits.
|
|
|
Copyright, credits and license:
|
|
* Contributed as supplement for Webware for Python and PyGreSQL
|
by Christoph Zwerschke in September 2005
|
* Based on the code of DBPool, contributed to Webware for Python
|
by Dan Green in December 2000
|
|
Licensed under the MIT license.
|
"""
|
|
from threading import Condition
|
|
from . import __version__
|
from .steady_db import connect
|
|
|
class PooledDBError(Exception):
|
"""General PooledDB error."""
|
|
|
class InvalidConnection(PooledDBError):
|
"""Database connection is invalid."""
|
|
|
class NotSupportedError(PooledDBError):
|
"""DB-API module not supported by PooledDB."""
|
|
|
class TooManyConnections(PooledDBError):
|
"""Too many database connections were opened."""
|
|
|
class PooledDB:
|
"""Pool for DB-API 2 connections.
|
|
After you have created the connection pool, you can use
|
connection() to get pooled, steady DB-API 2 connections.
|
"""
|
|
version = __version__
|
|
def __init__(
|
self, creator, mincached=0, maxcached=0,
|
maxshared=0, maxconnections=0, blocking=False,
|
maxusage=None, setsession=None, reset=True,
|
failures=None, ping=1,
|
*args, **kwargs):
|
"""Set up the DB-API 2 connection pool.
|
|
creator: either an arbitrary function returning new DB-API 2
|
connection objects or a DB-API 2 compliant database module
|
mincached: initial number of idle connections in the pool
|
(0 means no connections are made at startup)
|
maxcached: maximum number of idle connections in the pool
|
(0 or None means unlimited pool size)
|
maxshared: maximum number of shared connections
|
(0 or None means all connections are dedicated)
|
When this maximum number is reached, connections are
|
shared if they have been requested as shareable.
|
maxconnections: maximum number of connections generally allowed
|
(0 or None means an arbitrary number of connections)
|
blocking: determines behavior when exceeding the maximum
|
(if this is set to true, block and wait until the number of
|
connections decreases, otherwise an error will be reported)
|
maxusage: maximum number of reuses of a single connection
|
(0 or None means unlimited reuse)
|
When this maximum usage number of the connection is reached,
|
the connection is automatically reset (closed and reopened).
|
setsession: optional list of SQL commands that may serve to prepare
|
the session, e.g. ["set datestyle to ...", "set time zone ..."]
|
reset: how connections should be reset when returned to the pool
|
(False or None to rollback transcations started with begin(),
|
True to always issue a rollback for safety's sake)
|
failures: an optional exception class or a tuple of exception classes
|
for which the connection failover mechanism shall be applied,
|
if the default (OperationalError, InternalError) is not adequate
|
ping: determines when the connection should be checked with ping()
|
(0 = None = never, 1 = default = whenever fetched from the pool,
|
2 = when a cursor is created, 4 = when a query is executed,
|
7 = always, and all other bit combinations of these values)
|
args, kwargs: the parameters that shall be passed to the creator
|
function or the connection constructor of the DB-API 2 module
|
"""
|
try:
|
threadsafety = creator.threadsafety
|
except AttributeError:
|
try:
|
if not callable(creator.connect):
|
raise AttributeError
|
except AttributeError:
|
threadsafety = 2
|
else:
|
threadsafety = 0
|
if not threadsafety:
|
raise NotSupportedError("Database module is not thread-safe.")
|
self._creator = creator
|
self._args, self._kwargs = args, kwargs
|
self._blocking = blocking
|
self._maxusage = maxusage
|
self._setsession = setsession
|
self._reset = reset
|
self._failures = failures
|
self._ping = ping
|
if mincached is None:
|
mincached = 0
|
if maxcached is None:
|
maxcached = 0
|
if maxconnections is None:
|
maxconnections = 0
|
if maxcached:
|
if maxcached < mincached:
|
maxcached = mincached
|
self._maxcached = maxcached
|
else:
|
self._maxcached = 0
|
if threadsafety > 1 and maxshared:
|
self._maxshared = maxshared
|
self._shared_cache = [] # the cache for shared connections
|
else:
|
self._maxshared = 0
|
if maxconnections:
|
if maxconnections < maxcached:
|
maxconnections = maxcached
|
if maxconnections < maxshared:
|
maxconnections = maxshared
|
self._maxconnections = maxconnections
|
else:
|
self._maxconnections = 0
|
self._idle_cache = [] # the actual pool of idle connections
|
self._lock = Condition()
|
self._connections = 0
|
# Establish an initial number of idle database connections:
|
idle = [self.dedicated_connection() for i in range(mincached)]
|
while idle:
|
idle.pop().close()
|
|
def steady_connection(self):
|
"""Get a steady, unpooled DB-API 2 connection."""
|
return connect(
|
self._creator, self._maxusage, self._setsession,
|
self._failures, self._ping, True, *self._args, **self._kwargs)
|
|
def connection(self, shareable=True):
|
"""Get a steady, cached DB-API 2 connection from the pool.
|
|
If shareable is set and the underlying DB-API 2 allows it,
|
then the connection may be shared with other threads.
|
"""
|
if shareable and self._maxshared:
|
self._lock.acquire()
|
try:
|
while (not self._shared_cache and self._maxconnections
|
and self._connections >= self._maxconnections):
|
self._wait_lock()
|
if len(self._shared_cache) < self._maxshared:
|
# shared cache is not full, get a dedicated connection
|
try: # first try to get it from the idle cache
|
con = self._idle_cache.pop(0)
|
except IndexError: # else get a fresh connection
|
con = self.steady_connection()
|
else:
|
con._ping_check() # check this connection
|
con = SharedDBConnection(con)
|
self._connections += 1
|
else: # shared cache full or no more connections allowed
|
self._shared_cache.sort() # least shared connection first
|
con = self._shared_cache.pop(0) # get it
|
while con.con._transaction:
|
# do not share connections which are in a transaction
|
self._shared_cache.insert(0, con)
|
self._wait_lock()
|
self._shared_cache.sort()
|
con = self._shared_cache.pop(0)
|
con.con._ping_check() # check the underlying connection
|
con.share() # increase share of this connection
|
# put the connection (back) into the shared cache
|
self._shared_cache.append(con)
|
self._lock.notify()
|
finally:
|
self._lock.release()
|
con = PooledSharedDBConnection(self, con)
|
else: # try to get a dedicated connection
|
self._lock.acquire()
|
try:
|
while (self._maxconnections
|
and self._connections >= self._maxconnections):
|
self._wait_lock()
|
# connection limit not reached, get a dedicated connection
|
try: # first try to get it from the idle cache
|
con = self._idle_cache.pop(0)
|
except IndexError: # else get a fresh connection
|
con = self.steady_connection()
|
else:
|
con._ping_check() # check connection
|
con = PooledDedicatedDBConnection(self, con)
|
self._connections += 1
|
finally:
|
self._lock.release()
|
return con
|
|
def dedicated_connection(self):
|
"""Alias for connection(shareable=False)."""
|
return self.connection(False)
|
|
def unshare(self, con):
|
"""Decrease the share of a connection in the shared cache."""
|
self._lock.acquire()
|
try:
|
con.unshare()
|
shared = con.shared
|
if not shared: # connection is idle,
|
try: # so try to remove it
|
self._shared_cache.remove(con) # from shared cache
|
except ValueError:
|
pass # pool has already been closed
|
finally:
|
self._lock.release()
|
if not shared: # connection has become idle,
|
self.cache(con.con) # so add it to the idle cache
|
|
def cache(self, con):
|
"""Put a dedicated connection back into the idle cache."""
|
self._lock.acquire()
|
try:
|
if not self._maxcached or len(self._idle_cache) < self._maxcached:
|
con._reset(force=self._reset) # rollback possible transaction
|
# the idle cache is not full, so put it there
|
self._idle_cache.append(con) # append it to the idle cache
|
else: # if the idle cache is already full,
|
con.close() # then close the connection
|
self._connections -= 1
|
self._lock.notify()
|
finally:
|
self._lock.release()
|
|
def close(self):
|
"""Close all connections in the pool."""
|
self._lock.acquire()
|
try:
|
while self._idle_cache: # close all idle connections
|
con = self._idle_cache.pop(0)
|
try:
|
con.close()
|
except Exception:
|
pass
|
if self._maxshared: # close all shared connections
|
while self._shared_cache:
|
con = self._shared_cache.pop(0).con
|
try:
|
con.close()
|
except Exception:
|
pass
|
self._connections -= 1
|
self._lock.notifyAll()
|
finally:
|
self._lock.release()
|
|
def __del__(self):
|
"""Delete the pool."""
|
try:
|
self.close()
|
except Exception:
|
pass
|
|
def _wait_lock(self):
|
"""Wait until notified or report an error."""
|
if not self._blocking:
|
raise TooManyConnections
|
self._lock.wait()
|
|
|
# Auxiliary classes for pooled connections
|
|
class PooledDedicatedDBConnection:
|
"""Auxiliary proxy class for pooled dedicated connections."""
|
|
def __init__(self, pool, con):
|
"""Create a pooled dedicated connection.
|
|
pool: the corresponding PooledDB instance
|
con: the underlying SteadyDB connection
|
"""
|
# basic initialization to make finalizer work
|
self._con = None
|
# proper initialization of the connection
|
if not con.threadsafety():
|
raise NotSupportedError("Database module is not thread-safe.")
|
self._pool = pool
|
self._con = con
|
|
def close(self):
|
"""Close the pooled dedicated connection."""
|
# Instead of actually closing the connection,
|
# return it to the pool for future reuse.
|
if self._con:
|
self._pool.cache(self._con)
|
self._con = None
|
|
def __getattr__(self, name):
|
"""Proxy all members of the class."""
|
if self._con:
|
return getattr(self._con, name)
|
else:
|
raise InvalidConnection
|
|
def __del__(self):
|
"""Delete the pooled connection."""
|
try:
|
self.close()
|
except Exception:
|
pass
|
|
|
class SharedDBConnection:
|
"""Auxiliary class for shared connections."""
|
|
def __init__(self, con):
|
"""Create a shared connection.
|
|
con: the underlying SteadyDB connection
|
"""
|
self.con = con
|
self.shared = 1
|
|
def __lt__(self, other):
|
if self.con._transaction == other.con._transaction:
|
return self.shared < other.shared
|
else:
|
return not self.con._transaction
|
|
def __le__(self, other):
|
if self.con._transaction == other.con._transaction:
|
return self.shared <= other.shared
|
else:
|
return not self.con._transaction
|
|
def __eq__(self, other):
|
return (self.con._transaction == other.con._transaction
|
and self.shared == other.shared)
|
|
def __ne__(self, other):
|
return not self.__eq__(other)
|
|
def __gt__(self, other):
|
return other.__lt__(self)
|
|
def __ge__(self, other):
|
return other.__le__(self)
|
|
def share(self):
|
"""Increase the share of this connection."""
|
self.shared += 1
|
|
def unshare(self):
|
"""Decrease the share of this connection."""
|
self.shared -= 1
|
|
|
class PooledSharedDBConnection:
|
"""Auxiliary proxy class for pooled shared connections."""
|
|
def __init__(self, pool, shared_con):
|
"""Create a pooled shared connection.
|
|
pool: the corresponding PooledDB instance
|
con: the underlying SharedDBConnection
|
"""
|
# basic initialization to make finalizer work
|
self._con = None
|
# proper initialization of the connection
|
con = shared_con.con
|
if not con.threadsafety() > 1:
|
raise NotSupportedError("Database connection is not thread-safe.")
|
self._pool = pool
|
self._shared_con = shared_con
|
self._con = con
|
|
def close(self):
|
"""Close the pooled shared connection."""
|
# Instead of actually closing the connection,
|
# unshare it and/or return it to the pool.
|
if self._con:
|
self._pool.unshare(self._shared_con)
|
self._shared_con = self._con = None
|
|
def __getattr__(self, name):
|
"""Proxy all members of the class."""
|
if self._con:
|
return getattr(self._con, name)
|
else:
|
raise InvalidConnection
|
|
def __del__(self):
|
"""Delete the pooled connection."""
|
try:
|
self.close()
|
except Exception:
|
pass
|
