Python MySQLdb 模块，connect() 实例源码

我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用MySQLdb.connect()。

项目：BDA 作者：kotobukki | 项目源码 | 文件源码

def checkUserDbGrants(self, username):
        if self.cursor is None:
            self.connect()
        self.cursor.execute("select Drop_priv, Grant_priv, References_priv,Create_tmp_table_priv,"
                            "Lock_tables_priv,Create_routine_priv,Alter_routine_priv,Execute_priv,"
                            "Event_priv,Trigger_priv from mysql.db "
                            "where User ='%s'" % username)
        res = self.cursor.fetchall()
        flag = 0
        for row in res:
            # print row
            for key, value in row.items():
                if value == 'Y':
                    flag = 1
                    LOg.log_warn('setting %s = %s is suggested!' % (key, 'N'))
        if (flag == 0):
            Log.log_pass('All the setting are approriate!')

项目：code 作者：ActiveState | 项目源码 | 文件源码

def createDefCon(self):

                try:
                        host = "127.0.0.1"
                        port = 3306         ### default mysql port, change if you know better
                        user = "krisk"      ### def parameters
                        passwd = "kish"     ### def parameters
                        db = "loginfo"      ### connection.user_info contains the autho users

                        ### Create a connection object, use it to create a cursor

                        con = MySQLdb.connect(host = host  ,port = port , user = user,passwd = passwd ,db = db)
                        return con ### returns a connection object

                except: 
                        return 0;



####################################### Test connection #######################################################

项目：code 作者：ActiveState | 项目源码 | 文件源码

def populate_store(self, store):

        try:
            connection = None
            connection = MySQLdb.connect('localhost', 'annon', 'pass')
            cursor = connection.cursor()
            cursor.execute("Select * From `INFORMATION_SCHEMA`.`SCHEMATA`")
            rows = cursor.fetchall()

            for row in rows:
                store.append([row[0], row[1], row[2], row[3]])

        except MySQLdb.Error, e:
            store.append([str(e.args[0]), e.args[1], '', ''])

        finally:
            if connection != None:
                connection.close()

项目：supremm 作者：ubccr | 项目源码 | 文件源码

def getdbconnection(configsection, as_dict=False, defaultargs={}):
    """ Helper function that gets a database connection object from a config dictionary """

    dbengine = configsection['dbengine'] if 'dbengine' in configsection else 'MySQLDB'

    if dbengine == 'MySQLDB':

        dbargs = defaultargs.copy()
        # Convert the external configuration names to python PEP-249 config names
        translate = {"host": "host", 
                     "defaultsfile": "read_default_file",
                     "user": "user",
                     "pass": "passwd",
                     "port": "port"}

        for confval, myval in translate.iteritems():
            if confval in configsection:
                dbargs[myval] = configsection[confval]

        if as_dict:
            dbargs['cursorclass'] = MySQLdb.cursors.DictCursor

        return MySQLdb.connect(**dbargs)
    else:
        raise Exception("Unsupported database engine %s" % (dbengine))

项目：pykit 作者：baishancloud | 项目源码 | 文件源码

def new_connection(conn_argkw, conv=None, options=None):

    # useful arg could be added in future.:
    # conn_argkw.init_command

    conv = conv or {}
    options = options or {}

    opt = {
        'autocommit': 1,
    }
    opt.update(options)

    conn = MySQLdb.connect(conv=conv, **conn_argkw)
    for k, v in opt.items():
        conn.query('set {k}={v}'.format(k=k, v=v))

    return conn

项目：health-mosconi 作者：GNUHealth-Mosconi | 项目源码 | 文件源码

def cursor(self, autocommit=False, readonly=False):
        conv = MySQLdb.converters.conversions.copy()
        conv[float] = lambda value, _: repr(value)
        conv[MySQLdb.constants.FIELD_TYPE.TIME] = MySQLdb.times.Time_or_None
        args = {
            'db': self.database_name,
            'sql_mode': 'traditional,postgresql',
            'use_unicode': True,
            'charset': 'utf8',
            'conv': conv,
        }
        uri = parse_uri(config.get('database', 'uri'))
        assert uri.scheme == 'mysql'
        if uri.hostname:
            args['host'] = uri.hostname
        if uri.port:
            args['port'] = uri.port
        if uri.username:
            args['user'] = uri.username
        if uri.password:
            args['passwd'] = urllib.unquote_plus(uri.password)
        conn = MySQLdb.connect(**args)
        cursor = Cursor(conn, self.database_name)
        cursor.execute('SET time_zone = "+00:00"')
        return cursor

项目：health-mosconi 作者：GNUHealth-Mosconi | 项目源码 | 文件源码

def list(cursor):
        now = time.time()
        timeout = config.getint('session', 'timeout')
        res = Database._list_cache
        if res and abs(Database._list_cache_timestamp - now) < timeout:
            return res
        cursor.execute('SHOW DATABASES')
        res = []
        for db_name, in cursor.fetchall():
            try:
                database = Database(db_name).connect()
            except Exception:
                continue
            cursor2 = database.cursor()
            if cursor2.test():
                res.append(db_name)
                cursor2.close(close=True)
            else:
                cursor2.close()
                database.close()
        Database._list_cache = res
        Database._list_cache_timestamp = now
        return res

项目：envoy-perf 作者：envoyproxy | 项目源码 | 文件源码

def AuthorizeMachineAndConnectDB(db_instance_name, project, username, database,
                                 logfile):
  """This function authorizes a machine to connect to a DB.

  Args:
    db_instance_name: name of the DB instance.
    project: name of the project in which the db_instance belongs to.
    username: username that will be used to log in to the DB
    database: name of the database to connect to
    logfile: logfile for the gcloud command in shell
  Returns:
    Returns the connection after being connected to the DB.
  """
  hostname = GetInstanceIP(db_instance_name, project)

  password = utils.GetRandomPassword()
  sh_utils.RunGCloudService(["users", "set-password", username, "%",
                             "--instance", db_instance_name, "--password",
                             password], project=project,
                            service="sql", logfile=logfile)
  print "DB Usernames and passwords are set."
  connection = MySQLdb.connect(host=hostname, user=username,
                               passwd=password, db=database)
  return connection

项目：DockerSecurityResearch 作者：puyangsky | 项目源码 | 文件源码

def getImageName(_index):
    global MAX
    if _index > MAX:
        return "-1"

    conn = MySQLdb.connect(IP, DBUSERNAME, DBPWD, DBNAME)
    cursor = conn.cursor()
    sql = "select url from image where id = %d" % _index
    cursor.execute(sql)
    url = cursor.fetchone()
    print url
    if url != None:
        url = str(url[0])
        items = url.split('/')
        if len(items) != 7:
            return "#error#"
        name = items[-3] + '/' + items[-2]
        return name
    else:
        print "+++None return"
        return "#error#"

    conn.close()

项目：dpspider 作者：doupengs | 项目源码 | 文件源码

def __init__(self,host,user,password,db,charset='utf8',logFile=None,color=True,debug=4):
        '''
        :param host: <class str|host name>
        :param user: <class str|user name>
        :param password: <class str|password>
        :param db: <class str|database name>
        :param charset: default='utf8' <class str>
        :param logFile: default=None <class str>
        :param color: default=True <class bool>
        :param debug: default=4 <class int|0 NONE,1 [Error],2 [Error][WARING],3 [Error][WARING][INFO],4 ALL>
        '''
        self.logFile = logFile
        self.color = color
        self.debug = debug
        self.success = 0
        self.fail = 0
        self.repeat = 0
        self._conn = MYSQL.connect(host,user,password,db,charset=charset)
        self._cursor = self._conn.cursor()

项目：infra_helpers 作者：ayush-sharma | 项目源码 | 文件源码

def get_data_from_mysql(host: str, username: str, password: str, db: str, sql: str):
    """ Run SQL query and get data from MySQL table. """

    db = MySQLdb.connect(host, username, password, db)

    cursor = db.cursor(MySQLdb.cursors.DictCursor)

    try:
        cursor.execute(sql)
        data = cursor.fetchall()
    except Exception as e:
        print("MySQL error %s: %s" % (e.args[0], e.args[1]))
        data = None

    db.close()

    return data