python类ConnectionPool()的实例源码

def __init__(self, settings):
        dbname = settings['MYSQL_DBNAME']
        tbname = settings['TIEBA_NAME']
        if not dbname.strip():
            raise ValueError("No database name!")
        if not tbname.strip():
            raise ValueError("No tieba name!")          
        if isinstance(tbname, unicode):
            settings['TIEBA_NAME'] = tbname.encode('utf8')

        self.settings = settings

        self.dbpool = adbapi.ConnectionPool('MySQLdb',
            host=settings['MYSQL_HOST'],
            db=settings['MYSQL_DBNAME'],
            user=settings['MYSQL_USER'],
            passwd=settings['MYSQL_PASSWD'],
            charset='utf8mb4',
            cursorclass = MySQLdb.cursors.DictCursor,
            init_command = 'set foreign_key_checks=0' #??????
        )

def from_settings(cls, settings):
        dbparms = dict(
            host = settings["MYSQL_HOST"],
            db = settings["MYSQL_DBNAME"],
            user = settings["MYSQL_USER"],
            passwd = settings["MYSQL_PASSWORD"],
            charset='utf8',
            cursorclass=MySQLdb.cursors.DictCursor,
            use_unicode=True,
        )
        dbpool = adbapi.ConnectionPool("MySQLdb", **dbparms)

        return cls(dbpool)

def from_settings(cls, settings):
        dbparms = dict(
            host=settings["MYSQL_HOST"],
            db=settings["MYSQL_DBNAME"],
            user=settings["MYSQL_USER"],
            passwd=settings["MYSQL_PASSWORD"],
            charset='utf8',
            cursorclass=MySQLdb.cursors.DictCursor,
            use_unicode=True,
        )
        dbpool = adbapi.ConnectionPool("MySQLdb", **dbparms)

        return cls(dbpool)

def from_settings(cls, settings):
        dbparms = dict(
            host=settings["MYSQL_HOST"],
            db=settings["MYSQL_DBNAME"],
            user=settings["MYSQL_USER"],
            passwd=settings["MYSQL_PASSWORD"],
            charset='utf8',
            cursorclass=MySQLdb.cursors.DictCursor,
            use_unicode=True,
        )
        dbpool = adbapi.ConnectionPool("MySQLdb", **dbparms)

        return cls(dbpool)

def from_settings(cls, settings):
        dbparms = dict(
            host=settings["MYSQL_HOST"],
            db=settings["MYSQL_DBNAME"],
            user=settings["MYSQL_USER"],
            passwd=settings["MYSQL_PASSWORD"],
            charset='utf8',
            cursorclass=MySQLdb.cursors.DictCursor,
            use_unicode=True,
        )
        dbpool = adbapi.ConnectionPool("MySQLdb", **dbparms)

        return cls(dbpool)

def from_settings(cls, settings):
        dbparms = dict(
            host = settings["MYSQL_HOST"],
            db = settings["MYSQL_DBNAME"],
            user = settings["MYSQL_USER"],
            password = settings["MYSQL_PASSWORD"],
            charset = "utf8",
            cursorclass = MySQLdb.cursors.DictCursor,
            use_unicode = True,
        )
        dbpool = adbapi.ConnectionPool("MySQLdb", **dbparms)
        return cls(dbpool)

def __init__(self, info):
        self.info = info
        self.dbpool = adbapi.ConnectionPool(**self.info)

def __setstate__(self, state):
        self.__dict__ = state
        self.info['password'] = getpass.getpass('Database password for %s: ' % (self.info['user'],))
        self.dbpool = adbapi.ConnectionPool(**self.info)
        del self.info['password']

def from_settings(cls,settings):
        dbparms = dict(
            host = settings["MYSQL_HOST"],
            port = settings["MYSQL_PORT"],
            user = settings["MYSQL_USER"],
            passwd = settings["MYSQL_PASSWD"],
            db = settings["MYSQL_DB"],
            use_unicode = True,
            charset="utf8",
        )
        dbpool = adbapi.ConnectionPool("pymysql",**dbparms)
        return cls(dbpool)

def set_up_client():
    # set up client with:
    #
    # set Twisted debug level
    # connect to the db
    global dbconn, dbd, dbpool
    DEBUG_MODE = True
    setDebugging(DEBUG_MODE)
    log.startLogging(sys.stdout)
    # db connection and deferred
    dbconn = txpostgres.Connection()
    dbd = dbconn.connect("dbname=news user=user password=test1 host=127.0.0.1 application_name='news-collect'")
    dbpool = adbapi.ConnectionPool('psycopg2', dbname='news', user='user', password='test1')

def __init__(self):
        dbargs = dict(
             host = '127.0.0.1',
             db = 'test',
             user = 'root',
             passwd = '123456',
             cursorclass = MySQLdb.cursors.DictCursor,
             charset = 'utf8',
             use_unicode = True
            )
        self.dbpool = adbapi.ConnectionPool('MySQLdb',**dbargs)

def from_crawler(cls, crawler):
        settings = crawler.settings
        kw = dict(
            host=settings.get('MYSQL_HOST',' localhost'),
            port=settings.get('MYSQL_PORT', 3306),
            user=settings.get('MYSQL_USER', 'root'),
            db=settings.get('MYSQL_DB', 'test'),
            passwd=settings.get('MYSQL_PASSWD'),
            charset='utf8',
            use_unicode=True,
        )
        dbpool = adbapi.ConnectionPool ('MySQLdb', **kw)
        return cls (dbpool)

def from_settings(cls, settings):
        dbparams = dict(
            host=settings["MYSQL_HOST"],
            port=settings["MYSQL_PORT"],
            user=settings["MYSQL_USER"],
            passwd=settings["MYSQL_PWD"],
            db=settings["MYSQL_DB"],
            charset=settings["MYSQL_CHARSET"],
            use_unicode=settings["MYSQL_USER_UNICODE"],
            cursorclass=MySQLdb.cursors.DictCursor,
        )
        # ????adbapi????dbpool????MyTwistedPipeline?????
        dbpool = adbapi.ConnectionPool("MySQLdb", **dbparams)
        return cls(dbpool)

def __init__(self):
        self.dbpool = adbapi.ConnectionPool('pymysql',host='127.0.0.1', db='hospitalinfo',
                user='root', passwd='', cursorclass=pymysql.cursors.DictCursor,
                charset='utf8', use_unicode=True)

def __init__(self,dbpool):
        self.dbpool=dbpool
        ''' ?????????????????????????settings????????????
            self.dbpool=adbapi.ConnectionPool('MySQLdb',
                                          host='127.0.0.1',
                                          db='crawlpicturesdb',
                                          user='root',
                                          passwd='123456',
                                          cursorclass=MySQLdb.cursors.DictCursor,
                                          charset='utf8',
                                          use_unicode=False)'''

def __init__(self):
        self.dbpool = adbapi.ConnectionPool('MySQLdb',
            db = 'guazi',
            user = 'root',
            passwd = '',
            cursorclass = MySQLdb.cursors.DictCursor,
            charset = 'utf8',
            use_unicode = True
            )

def __init__(self):
        self.dbpool=adbapi.ConnectionPool('MySQLdb',
                host='127.0.0.1',
                db='liaohong_test',
                user='root',
                passwd='liaohong',
                cursorclass=MySQLdb.cursors.DictCursor,
                charset='utf8',
                use_unicode=True)

def from_settings(cls, settings):
        dbparms = config.DB_config.get("mysql")
        dbparms['db'] = config.database
        dbparms['cursorclass'] = MySQLdb.cursors.DictCursor
        dbparms['use_unicode'] = True

        dbpool = adbapi.ConnectionPool("MySQLdb", **dbparms)

        return cls(dbpool)

def __init__(self, info):
        self.info = info
        self.dbpool = adbapi.ConnectionPool(**self.info)

def __setstate__(self, state):
        self.__dict__ = state
        self.info['password'] = getpass.getpass('Database password for %s: ' % (self.info['user'],))
        self.dbpool = adbapi.ConnectionPool(**self.info)
        del self.info['password']

def __init__(self):    
        dbargs = settings.get('DB_CONNECT')    
        db_server = settings.get('DB_SERVER')    
        dbpool = adbapi.ConnectionPool(db_server, **dbargs)    
        self.dbpool = dbpool

def from_settings(cls, settings):
        dbparams = dict(
            host=settings['MYSQL_HOST'],
            port=settings['MYSQL_PORT'],
            db=settings['MYSQL_DBNAME'],
            user=settings['MYSQL_USER'],
            passwd=settings['MYSQL_PASSWD'],
            charset=settings['MYSQL_CHARSET'],
            cursorclass=MySQLdb.cursors.DictCursor,
            use_unicode=False,
        )
        dbpool = adbapi.ConnectionPool('MySQLdb', **dbparams)  # **?????????????
        return cls(dbpool)

    # pipeline????

def start(self):
        """Create the ADBAPI connection pool.

        """
        Registry.DBPOOL = adbapi.ConnectionPool('psycopg2', host=self.host,
                  user=self.user, password=self.password,
                  database=self.database)

def from_settings(cls,settings):
        dbparams=dict(
            host=settings['MYSQL_HOST'],#??settings????
            db=settings['MYSQL_DBNAME'],
            user=settings['MYSQL_USER'],
            passwd=settings['MYSQL_PASSWD'],
            charset='utf8',#??????????????????
            cursorclass=MySQLdb.cursors.DictCursor,
            use_unicode=False,
        )
        dbpool=adbapi.ConnectionPool('MySQLdb',**dbparams)#**?????????????,???host=xxx,db=yyy....
        return cls(dbpool)#???dbpool???????self?????

def run_conn_fun(fun,*args):
    try:
        global conn
        result=yield  getattr(conn,fun)(*args)
    except (MySQLdb.OperationalError,adbapi.ConnectionLost):
        try:
            conn.close()
        except:
            pass
        conn=adbapi.ConnectionPool("MySQLdb",host=MYSQLCONFIG["host"],user=MYSQLCONFIG["user"],passwd=MYSQLCONFIG["passwd"],charset=MYSQLCONFIG["charset"],port=MYSQLCONFIG["port"],db=MYSQLCONFIG["db"],cp_reconnect=True)
        result=yield  getattr(conn,fun)(*args)
    defer.returnValue(result)

def from_settings(cls, settings):
        dbargs = dict(
            host=settings['MYSQL_HOST'],
            db=settings['MYSQL_DBNAME'],
            user=settings['MYSQL_USER'],
            passwd=settings['MYSQL_PASSWD'],
            port=settings['MYSQL_PORT'],
            charset='utf8',
            use_unicode=True,
        )
        dbpool = adbapi.ConnectionPool('MySQLdb', **dbargs)
        return cls(dbpool)

def from_settings(cls, settings):
        dbparms = dict(
            host = settings["MYSQL_HOST"],
            db = settings["MYSQL_DBNAME"],
            user = settings["MYSQL_USER"],
            password = settings["MYSQL_PASSWORD"],
            charset = 'utf8',
            cursorclass = MySQLdb.cursors.DictCursor,
            use_unicode = True,
        )
        dbpool = adbapi.ConnectionPool("MySQLdb",**dbparms)
        return cls(dbpool)

def opened_spider(self, spider):
        self.db = adbapi.ConnectionPool(**self.config)

def from_settings(cls, settings):
        dbparms = dict(
            host = settings["MYSQL_HOST"],
            db = settings["MYSQL_DBNAME"],
            user = settings["MYSQL_USER"],
            passwd = settings["MYSQL_PASSWORD"],
            charset='utf8',
            cursorclass=MySQLdb.cursors.DictCursor,
            use_unicode=True,
        )
        dbpool = adbapi.ConnectionPool("MySQLdb", **dbparms)

        return cls(dbpool)

def from_settings(cls, settings):
        dbparms = dict(
            host=settings["MYSQL_HOST"],
            db=settings["MYSQL_DBNAME"],
            user=settings["MYSQL_USER"],
            passwd=settings["MYSQL_PASSWORD"],
            charset='utf8',
            cursorclass=MySQLdb.cursors.DictCursor,
            use_unicode=True,
        )
        dbpool = adbapi.ConnectionPool("MySQLdb", **dbparms)

        return cls(dbpool)