Python连接MySQL、PostgreSQL数据库实现过程

>>> pip3 install pymysql,psycopg2
>>> #多环境用户需要主要pip版本

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

>>> # X.X 为 PyMySQL 的版本号
>>> curl -L https://github.com/PyMySQL/PyMySQL/tarball/pymysql-X.X | tar xz
>>> cd PyMySQL*
>>> python3 setup.py install
>>> # 现在你可以删除 PyMySQL* 目录

database – 数据库名称

user – 用户名

password – 密码

host – 服务器地址 (如果不提供默认连接Unix Socket)

port – 连接端口 (默认5432)

#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文件
#导入所需包
import os
import pymysql
import psycopg2
from common.load_ini import load_ini

#拼接路径以列表的形式进行读取配置文件
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']
#对库信息进行赋值，此方法只作为演示，大佬可自行编写最优写法
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数据库函数
def my_sql_database():
    #连库相关信息
    mydb = pymysql.connect(
        host=mysql_host,
        port=mysql_port,
        user=mysql_username,
        passwd=mysql_password,
        db=mysql_database,
        charset='utf8'
    )
    mycursor = mydb.cursor()
    # 写入SQL语句对数据库进行增删改操作即可
    sql = "select * from jg_staff where staff_name = '测试1华北项目组'"
    # mycursor.execute("SHOW DATABASES")
    mycursor.execute(sql)
    print(mycursor.fetchall())
    mydb.commit()  # 查询时不需需调用，此方法用于提交当前事务。如果不调用这个方法，无论做了什么修改，自从上次调用#commit()是不可见的

    # 关闭数据库
    mycursor.close()  # 关闭游标
    mycursor.close()  # 关闭数据库连接

#连接PostgreSQL数据库函数
def pg_sql_database():
	#PostgreSQL数据库连接方法和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()) #打印结果
    conn.commit()  # 查询时不需需调用，此方法用于提交当前事务。如果不调用这个方法，无论做了什么修改，自从上次调用#commit()是不可见的
    # 关闭数据库
    cur.close()  # 关闭游标
    cur.close()  # 关闭数据库连接

def my_sql_insert(sql_case):
    """
    往sql插入数据
    :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的列转换为浮点数
                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以数据类型输出
    :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