使用前需要先用pip安装以下包

Mysql

pip install PyMySQL

Postgresql

pip install psycopg2

Oracle

pip install cx_Oracle 

案例代码

# coding=UTF-8

import cx_Oracle
import psycopg2
import pymysql


# mysql连接
def mysql(instancetype, host, port, database, user, password, charset, sql):
    """
    host        连接地址\n
    port        端口\n
    database    数据库名\n
    user        用户名\n
    password    密码\n
    charset     编码\n
    sql         SQL语句\n
    """
    try:
        # 打开数据库连接
        db = pymysql.connect(host=host,
                             user=user,
                             password=password,
                             port=int(port),
                             database=database,
                             charset=charset)
        try:
            # 使用 cursor() 方法创建一个游标对象 cursor
            cursor = db.cursor()
            # 使用 execute()  方法执行 SQL 查询
            cursor.execute(sql)
            # 使用 fetchone() 方法获取单条数据.
            result = cursor.fetchall()
            print(result)
        except Exception as e:
            # 获取数据异常
            print("Error: ", e)
        finally:
            # 关闭数据库连接
            db.close()
    except Exception as e:
        # 数据库连接异常
        print("Error: ", e)
    pass


# postgresql连接
def postgresql(instancetype, host, port, database, user, password, charset, sql):
    """
    host        连接地址\n
    port        端口\n
    database    数据库名\n
    user        用户名\n
    password    密码\n
    charset     编码\n
    sql         SQL语句\n
    """
    try:
        # 打开数据库连接
        db = psycopg2.connect(host=host,
                              user=user,
                              password=password,
                              port=int(port),
                              database=database,
                              client_encoding=charset)

        try:
            # 使用 cursor() 方法创建一个游标对象 cursor
            cursor = db.cursor()
            # 使用 execute()  方法执行 SQL 查询
            cursor.execute(sql)
            # 使用 fetchone() 方法获取单条数据.
            result = cursor.fetchall()
            print(result)
        except Exception as e:
            # 获取数据异常
            print("Error: ", e)
        finally:
            # 关闭数据库连接
            db.close()
    except Exception as e:
        # 数据库连接异常
        print("Error: ", e)
    pass


def oracle(instancetype, host, port, database, user, password, charset, sql):
    try:
        user = user
        password = password
        host = host + ":" + port
        service_name = database
        conn_str = f"{user}/{password}@{host}/{service_name}"
        connect = cx_Oracle.connect(conn_str)
        try:
            cursor = connect.cursor()
            cursor.execute(sql)
            result = cursor.fetchall()
            print(result)
        except Exception as e:
            print("Error1: ", e)
        finally:
            connect.close()
    except Exception as e:
        print("Error2: ", e)


def sql_client(instancetype, sqltype, host, port, database, user, password, charset, sql):
    if sqltype == "mysql":
        print("mysql......")
        mysql(instancetype, host, port, database, user, password, charset, sql)
    elif sqltype == "postgresql":
        print("postgresql......")
        postgresql(instancetype, host, port, database, user, password, charset, sql)
    elif sqltype == "oracle":
        print("oracle......")
        oracle(instancetype, host, port, database, user, password, charset, sql)
    else:
        print("数据库类型:" + sqltype + "未适配")
    pass

Last Updated on 2023 18 10 月