快捷導(dǎo)航

Python連接MySQL、PostgreSQL數(shù)據(jù)庫實(shí)現(xiàn)過程

更新時間：2025年12月15日 11:14:47 作者：回憶式~過去.

文章介紹了如何在Python中使用`pymysql`和`psycopg2`模塊連接MySQL和PostgreSQL數(shù)據(jù)庫,包括安裝庫、連接參數(shù)、數(shù)據(jù)庫操作和關(guān)閉連接的步驟

一、安裝庫

Python連接MySQL、PostgreSQL數(shù)據(jù)庫需要導(dǎo)入相關(guān)的模塊，分別是”pymysql“和”psycopg2“模塊，我們可以在Pycharm工具中直接搜索安裝對應(yīng)的庫然后導(dǎo)入即可，兩個數(shù)據(jù)庫連接方式基本一致，所以只拿其中一個作為演示。

如果工具安裝失敗(有可能是網(wǎng)絡(luò)原因)，也可以使用pip命令進(jìn)行安裝

>>> pip3 install pymysql,psycopg2
>>> #多環(huán)境用戶需要主要pip版本

如果你的系統(tǒng)不支持 pip 命令，可以使用以下方式安裝：

1、使用 git 命令下載安裝包安裝(你也可以手動下載)：

>>> git clone https://github.com/PyMySQL/PyMySQL
>>> cd PyMySQL/
>>> python3 setup.py install

2、如果需要制定版本號，可以使用 curl 命令來安裝：

>>> # X.X 為 PyMySQL 的版本號
>>> curl -L https://github.com/PyMySQL/PyMySQL/tarball/pymysql-X.X | tar xz
>>> cd PyMySQL*
>>> python3 setup.py install
>>> # 現(xiàn)在你可以刪除 PyMySQL* 目錄

二、連接數(shù)據(jù)庫

連接數(shù)據(jù)庫會用到的參數(shù)：

database – 數(shù)據(jù)庫名稱

user – 用戶名

password – 密碼

host – 服務(wù)器地址 (如果不提供默認(rèn)連接Unix Socket)

port – 連接端口 (默認(rèn)5432)

用pymysql模塊中的connect連接數(shù)據(jù)庫，同時用cursor()函數(shù)創(chuàng)建游標(biāo)，用于接收返回的結(jié)果。

這里返回了一個游標(biāo)實(shí)例對象，說明你已經(jīng)連接成功了。

三、數(shù)據(jù)庫操作

接下來我們可以使用 .execute 對數(shù)據(jù)庫進(jìn)行一系列的增刪改查操作

運(yùn)行成功后就可以在電腦上的數(shù)據(jù)庫中查找到相應(yīng)的變化啦，最后記得關(guān)閉數(shù)據(jù)庫的連接，以下是查詢結(jié)果：

四、代碼奉上

我是通過讀取配置文件讀取數(shù)據(jù)庫信息進(jìn)行連接的，一并付上代碼，有需要的可以參考，不需要的直接忽略

#setting.ini配置文件
[mysql]
# MySQL配置
MYSQL_HOST = 39.101.000.0
MYSQL_PORT = 5432
MYSQL_USER = test
MYSQL_PASSWD = xxxx
MYSQL_DB = xxxx_dev

[postgresql]
# postgresql配置
PG_HOST = 39.101.000.0
PG_PORT = 5432
PG_USER = test
PG_PASSWD = xxxxx
PG_DB = xxxxxx_dev

# load_ini.py文件，讀取ini配置文件方法
def load_ini(file_path):
    log.info("加載 {} 文件......".format(file_path))
    config = MyConfigParser()
    config.read(file_path, encoding="UTF-8")
    data = dict(config._sections)
    return data

#run.py文件
#導(dǎo)入所需包
import os
import pymysql
import psycopg2
from common.load_ini import load_ini

#拼接路徑以列表的形式進(jìn)行讀取配置文件
BASE_PATH = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
data_file_path = os.path.join(BASE_PATH, "config", "setting.ini")
mysql_data = load_ini(data_file_path)['mysql']
pg_data = load_ini(data_file_path)['postgresql']
#對庫信息進(jìn)行賦值，此方法只作為演示，大佬可自行編寫最優(yōu)寫法
mysql_host = mysql_data["MYSQL_HOST"]
mysql_port = int(mysql_data["MYSQL_PORT"])
mysql_username = mysql_data["MYSQL_USER"]
mysql_password = mysql_data["MYSQL_PASSWD"]
mysql_database = mysql_data["MYSQL_DB"]
pg_host = pg_data["PG_HOST"]
pg_port = int(pg_data["PG_PORT"])
pg_username = pg_data["PG_USER"]
pg_password = pg_data["PG_PASSWD"]
pg_database = pg_data["PG_DB"]

#連接MySQL數(shù)據(jù)庫函數(shù)
def my_sql_database():
    #連庫相關(guān)信息
    mydb = pymysql.connect(
        host=mysql_host,
        port=mysql_port,
        user=mysql_username,
        passwd=mysql_password,
        db=mysql_database,
        charset='utf8'
    )
    mycursor = mydb.cursor()
    # 寫入SQL語句對數(shù)據(jù)庫進(jìn)行增刪改操作即可
    sql = "select * from jg_staff where staff_name = '測試1華北項(xiàng)目組'"
    # mycursor.execute("SHOW DATABASES")
    mycursor.execute(sql)
    print(mycursor.fetchall())
    mydb.commit()  # 查詢時不需需調(diào)用，此方法用于提交當(dāng)前事務(wù)。如果不調(diào)用這個方法，無論做了什么修改，自從上次調(diào)用#commit()是不可見的

    # 關(guān)閉數(shù)據(jù)庫
    mycursor.close()  # 關(guān)閉游標(biāo)
    mycursor.close()  # 關(guān)閉數(shù)據(jù)庫連接

#連接PostgreSQL數(shù)據(jù)庫函數(shù)
def pg_sql_database():
	#PostgreSQL數(shù)據(jù)庫連接方法和MySQL方法基本一致
    conn = psycopg2.connect(
            host=pg_host,
            port=pg_port,
            user=pg_username,
            password=pg_password,
            database=pg_database
    )
    cur = conn.cursor()
    sql = "select area from b_lands_attribute_2021 bla where id = '20545521'"
    cur.execute(sql)
    print(cur.fetchall()) #打印結(jié)果
    conn.commit()  # 查詢時不需需調(diào)用，此方法用于提交當(dāng)前事務(wù)。如果不調(diào)用這個方法，無論做了什么修改，自從上次調(diào)用#commit()是不可見的
    # 關(guān)閉數(shù)據(jù)庫
    cur.close()  # 關(guān)閉游標(biāo)
    cur.close()  # 關(guān)閉數(shù)據(jù)庫連接

def my_sql_insert(sql_case):
    """
    往sql插入數(shù)據(jù)
    :param sql_case:
    :return:
    """
    mydb = pymysql.connect(
        host=mysql_host,
        port=mysql_port,
        user=mysql_username,
        passwd=mysql_password,
        db=mysql_database,
        charset='utf8mb4'
    )
    mycursor = mydb.cursor()
    mycursor.execute(sql_case)
    mydb.commit()
    mydb.close()

def mysql_select(sql_case):
    """
    查詢sql以列表形式輸出
    :param sql_case:
    :return:
    """
    try:
        connection = pymysql.connect(
        host=mysql_host,
        port=mysql_port,
        user=mysql_username,
        passwd=mysql_password,
        db=mysql_database,
        charset='utf8mb4'
    )

        with connection.cursor() as cursor:
            cursor.execute(sql_case)

            results = []
            for row in cursor.fetchall():
                # 將包含Decimal的列轉(zhuǎn)換為浮點(diǎn)數(shù)
                processed_row = [float(item) if isinstance(item, Decimal) else item for item in row]
                results.append(processed_row)

    except MySQLError as e:
        print(f"Error: {e}")
        results = []

    finally:
        if connection:
            connection.close()

    return results


def my_sql_select(sql_select):
    """
    查詢sql以數(shù)據(jù)類型輸出
    :param sql_select:
    :return:
    """
    mydb = pymysql.connect(
        host=mysql_host,
        port=mysql_port,
        user=mysql_username,
        passwd=mysql_password,
        db=mysql_database,
        charset='utf8mb4'
    )
    mycursor = mydb.cursor()
    mycursor.execute(sql_select)
    nested_list = [list(row) for row in mycursor.fetchall()]
    mydb.close()
    return nested_list