vlambda博客
学习文章列表

class PooledDatabase(object): def __init__(self, database, max_connections=20, stale_timeout=None, timeout=None, **kwargs): self._max_connections = make_int(max_connections) self._stale_timeout = make_int(stale_timeout) self._wait_timeout = make_int(timeout) if self._wait_timeout == 0: self._wait_timeout = float('inf')
 # Available / idle connections stored in a heap, sorted oldest first. self._connections = []
 # Mapping of connection id to PoolConnection. Ordinarily we would want # to use something like a WeakKeyDictionary, but Python typically won't # allow us to create weak references to connection objects. self._in_use = {}
 # Use the memory address of the connection as the key in the event the # connection object is not hashable. Connections will not get # garbage-collected, however, because a reference to them will persist # in "_in_use" as long as the conn has not been closed. self.conn_key = id
        super(PooledDatabase, self).__init__(database, **kwargs)
 def _connect(self): while True: try: # Remove the oldest connection from the heap. ts, conn = heapq.heappop(self._connections) key = self.conn_key(conn) except IndexError: ts = conn = None logger.debug('No connection available in pool.') break else: if self._is_closed(conn): # This connecton was closed, but since it was not stale # it got added back to the queue of available conns. We # then closed it and marked it as explicitly closed, so # it's safe to throw it away now. # (Because Database.close() calls Database._close()). logger.debug('Connection %s was closed.', key) ts = conn = None elif self._stale_timeout and self._is_stale(ts): # If we are attempting to check out a stale connection, # then close it. We don't need to mark it in the "closed" # set, because it is not in the list of available conns # anymore. logger.debug('Connection %s was stale, closing.', key) self._close(conn, True) ts = conn = None else: break
 if conn is None: if self._max_connections and ( len(self._in_use) >= self._max_connections): raise MaxConnectionsExceeded('Exceeded maximum connections.') conn = super(PooledDatabase, self)._connect() ts = time.time() - random.random() / 1000 key = self.conn_key(conn) logger.debug('Created new connection %s.', key)
 self._in_use[key] = PoolConnection(ts, conn, time.time()) return conn
 def _close(self, conn, close_conn=False): key = self.conn_key(conn) if close_conn: super(PooledDatabase, self)._close(conn) elif key in self._in_use: pool_conn = self._in_use.pop(key) if self._stale_timeout and self._is_stale(pool_conn.timestamp): logger.debug('Closing stale connection %s.', key) super(PooledDatabase, self)._close(conn) elif self._can_reuse(conn): logger.debug('Returning %s to pool.', key) heapq.heappush(self._connections, (pool_conn.timestamp, conn)) else: logger.debug('Closed %s.', key)

class PooledMySQLDatabase(PooledDatabase, MySQLDatabase): def _is_closed(self, conn): try: conn.ping(False) except: return True else: return False

heapq.heappush(self._connections, (pool_conn.timestamp, conn))

If you want to manage transactions separately, you can use the Database.connection_context() context manager.
>>> with db.connection_context():...  # db connection is open....  pass...>>> db.is_closed() # db connection is closed.TrueThe connection_context() method can also be used as a decorator:
@db.connection_context()def prepare_database(): # DB connection will be managed by the decorator, which opens # a connection, calls function, and closes upon returning. db.create_tables(MODELS) # Create schema. load_fixture_data(db)