我们从Python开源项目中，提取了以下43个代码示例，用于说明如何使用twisted.enterprise.adbapi.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"], 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): 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=pymysql.cursors.DictCursor, use_unicode=True, ) dbpool = adbapi.ConnectionPool("pymysql", **dbparms) return cls(dbpool)
def __init__(self): self.dbpool = adbapi.ConnectionPool( dbapiName = 'MySQLdb', host = 'localhost', db = 'sina_spider_db', user = 'root', passwd = 'gongli', cursorclass = MySQLdb.cursors.DictCursor, charset = 'utf8mb4', )
def __init__(self): self.dbpool = adbapi.ConnectionPool( dbapiName='MySQLdb', host=settings.DB_HOST, db=settings.DB, user=settings.DB_NAME, passwd=settings.DB_PASSWD, cursorclass= MySQLdb.cursors.DictCursor, charset = 'utf8', use_unicode = False )
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) # dbpool???__init__ # ??twisted?mysql????????
def from_settings(cls, settings): dbargs = dict( host=settings['MYSQL_HOST'], db=settings['MYSQL_DBNAME'], user=settings['MYSQL_USER'], passwd=settings['MYSQL_PASSWD'], charset='utf8', cursorclass = MySQLdb.cursors.DictCursor, use_unicode= True, ) dbpool = adbapi.ConnectionPool('MySQLdb', **dbargs) return cls(dbpool)
def __init__(self): self.dbpool = adbapi.ConnectionPool('MySQLdb', db='spiders', user='root', passwd='build.ns', cursorclass=MySQLdb.cursors.DictCursor, charset='utf8', use_unicode=True)
def adb_connect_db(db_type,**kwargs): ''' db_type-->"MySQLdb" ''' dbpool = adbapi.ConnectionPool(db_type, **kwargs) return dbpool
