DBUtils
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: # builtin Exceptions might not exist any more
pass
def __enter__(self):
"""Enter a runtime context for the connection."""
return self
def __exit__(self, *exc):
"""Exit a runtime context for the connection."""
self.close()
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: # 最大允许连接数 小于 pool中最大空闲连接数
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:
with self._lock:
# 保证线程安全
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()
con = PooledSharedDBConnection(self, con)
else: # try to get a dedicated connection
with self._lock:
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
return con