安裝庫pymysql

pip install pymysql

使用前引入

import pymysql

使用步驟如下：

1.建立連線

db_config = {
    'host': 'api.****.com', # 主機
    'user': '*****',       #使用者名稱
    'password': '123456',  #密碼
    'port': 3306,         #埠 3306
    'database':'*****'   #資料庫名
}
# 建立連線
conn = pymysql.connect(**db_config)

2.建立遊標

cursor=conn.cursor()

3.準備並執行sql語句

sql='select leave_amount from member where id=100'
cursor.execute(sql)

4.獲取查詢結果，類似於卸貨，查詢結果如果有10條，三個獲取結果語句均使用，則fetchone()獲得一條，fetchmany(3)獲得三條，fetchall()即獲得6條，三條語句合起來為查詢結果為10條，如果僅使用fetchall()則獲取10條

# 獲取一條查詢結果，以元組的形式返回
cursor.fetchone()
# 獲取三條查詢結果
cursor.fetchmany(3)
# 獲取全部查詢結果
cursor.fetchall()

　上述結果均以元組形式返回，獲取結果以字典形式返回

# 需要引入 DictCursor
from pymysql.cursors import DictCursor
# 建立遊標的時候，指定為DictCursor
cursor=conn.cursor(DictCursor)

查詢資料庫可以封裝起來，後續直接匯入使用

# 配置檔案中db_config
MYSQL_CONFIG = {
    'host': 'api.*****.com',
    'user': '*****',
    'password': '123456',
    'port': 3306,
    'database':'*****',
    'engine':'mysql',
    'autocommit':True    
}


import pymysql
from pymysql.cursors import DictCursor

class Handle_SQL:

    def __init__(self,db_config):
        # 建立資料庫連線
        # 根據不同的資料庫，建立不同的連結
        # 如果engine沒有值預設返回mysql
        engine = db_config.pop('engine', 'mysql')
        if engine.lower() == 'mysql':
            self.conn = pymysql.connect(**db_config)
        else:
            pass# 此處可以配置其他的資料庫

    def get_fetchone(self,sql,res_type = 't'):
        """
        返回一條資料
        :param sql:
        :param size: 預設返回元組
        :return:
        """
        # 建立遊標
        # 返回元組型別的資料
        if res_type == 't':
            with self.conn.cursor() as cursor:
                cursor.execute(sql)
                return cursor.fetchone()
        # 返回字典型別的資料
        else:
            with self.conn.cursor(DictCursor) as cursor:
                cursor.execute(sql)
                return cursor.fetchone()

    # 獲取多個數據
    def get_fetchmany(self,sql,size,res_type='t'):
        if res_type == 't':
            with self.conn.cursor() as cursor:
                cursor.execute(sql)
                return cursor.fetchmany(size)
        # 返回字典型別的資料
        else:
            with self.conn.cursor(DictCursor) as cursor:
                cursor.execute(sql)
                return cursor.fetchmany(size)

    # 獲取全部資料
    def get_fetchall(self,sql,res_type='t'):
        if res_type== 't':
            with self.conn.cursor() as cursor:
                cursor.execute(sql)
                return cursor.fetchall()
        # 返回字典型別的資料
        else:
            with self.conn.cursor(DictCursor) as cursor:
                cursor.execute(sql)
                return cursor.fetchall()

    def exist_sql(self,sql):
        if self.get_fetchone(sql):
            return True
        else:
            return False

    def __del__(self):
        self.conn.close()

本文來自部落格園，作者：大頭~~，轉載請註明原文連結：https://www.cnblogs.com/xiaoying-guo/p/15066361.html