当前位置：首页 > news >正文

python连接数据库

news 2025/7/2 14:38:30

python连接MYSQL、postgres、oracle等的基本操作

python连接mysql

MySQLdb

MySQLdb又叫MySQL-python ，是 Python 连接 MySQL 的一个驱动，很多框架都也是基于此库进行开发，只支持 Python2.x，而且安装的时候有很多前置条件，因为它是基于C开发的库，在 Windows 平台安装非常不友好，经常出现失败的情况，不推荐使用，取代的是它的衍生版本。

Mysqlclient

MySQLdb的 Fork 版本，完全兼容 MySQLdb，同时支持 Python3.x，是 Django ORM的依赖工具，如果你想使用原生 SQL 来操作数据库，那么推荐此驱动。Mysqlclient是一个C扩展模块,编译安装可能会导致报各种错误。
安装：pip install mysqlclient
如果出错，可以查看这个文档 https://blog.csdn.net/cn_1937/article/details/81533544

import MySQLdb
# 注意 这里需要额外导入, 不然会报module 'MySQLdb' has no attribute 'cursors'的错误
import MySQLdb.cursors as cors# 打开数据库连接 
db = MySQLdb.connect("localhost", "testuser", "test123", "TESTDB", charset='utf8',cursorclass=cors.DictCursor) 
# 使用cursor()方法获取操作游标  
cursor = db.cursor() 
# SQL 插入语句 
sql="insert into EMPLOYEE values(%s,%s,%s,%s)" 
try: # 执行sql语句 cursor.executemany(sql,[('Smith','Tom',15,'M',1500),('Mac', 'Mohan', 20, 'M', 2000)]) # 提交到数据库执行 db.commit() 
except: # Rollback in case there is any error db.rollback() 
# 关闭数据库连接 
db.close()

pymysql包

首先需要安装：pip install pymysql

import pymysqldb = pymysql.connect(host='localhost', user='root', password='mysql',  #连接到数据库需要指定的最少信息db = 'demo', database='demo', #指定选择的数据库，任选一个都可，前面是简写passwd='mysql', #密码也可以用passwd简写形式autocommit=True, #设置修改自动提交到数据库auth_plugin_map='mysql_native_password') #设置身份认证，8.0版本之后建议加上
cursor = db.cursor() #创建一个指针，之后基本所有的操作都是使用指针来实现
cursor.execute('show databases;') #执行SQL语句
db.commit() #将修改提交到数据库——如果连接时参数autocommit设置为false要加上
cursor.fetchall() #获取上一条SQL语句的执行结果，如查询结果等
cursor.fetchone() #获取执行结果的一行
db.close() #关闭数据库

常用的代码就是上面这几行，所以主要还是在使用SQL语句来实现功能。其中主要有两个注意事项：

autocommit参数：这个如果不设置的话，默认为false，那么使用execute函数执行SQL语句时就有数据库没有修改的风险，所以建议加上。
auth_plugin_map参数：这个也是一个不容易发现的bug，根据StackOverflow上的介绍，在8.0版本之后的MySQL
server，加上了一个身份验证的东西，和原来的不一样，如果不加上这个参数，可能会出现重启电脑，虽然服务在运行但是无法通过程序连接到数据库。所以保险起见，还是加上这个比较合适。

import pymysqltry:db = pymysql.connect(host='localhost', user='root', passwd='666666', port=3306, db='Mysql8')print('连接成功！')
except:print('something wrong!')
cursor = db.cursor()
cursor.execute("DROP TABLE IF EXISTS EMPLOYEE")
sql = """CREATE TABLE EMPLOYEE (FIRST_NAME  CHAR(20) NOT NULL,LAST_NAME  CHAR(20),AGE INT,  SEX CHAR(1),INCOME FLOAT )"""cursor.execute(sql)
print('建表成功！')
db.close()

获取操作数据库的异常

在操作数据库时，为了排除异常情况，往往会加上try … except …

try:c = cursor.execute(sql) #指针执行SQL语句
except Exception as e:self.logger.error(f"The SQL code Execution ERROR: "+str(e))return -1

这个句式就能捕捉到sql语句执行时的错误，debug更加方便。

多线程访问数据库 dbutils

当主程序是多线程同时在执行时，且每个线程都需要操作数据，如果只用pymysql实现，相当于一个连接得要一直开着，这样据说运行久了会出问题。因此可以采用dbutils这个第三方模块来实现数据库多连接。
这个第三方模块的原理就是构造一个数据库的连接池，如果有需要连接就直接去池中申请一个连接，然后使用完再释放。
使用前先安装：pip install dbutils，调用方式如下所示

from dbutils.pooled_db import PooledDB

创建连接池

PooledDB 基本参数：

mincached : 最少的空闲连接数，如果空闲连接数小于这个数，Pool自动创建新连接;
maxcached : 最大的空闲连接数，如果空闲连接数大于这个数，Pool则关闭空闲连接;
maxconnections : 最大的连接数;
blocking : 当连接数达到最大的连接数时，在请求连接的时候，如果这个值是True，请求连接的程序会一直等待，直到当前连接数小于最- 连接数，如果这个值是False，会报错;

def mysql_connection():maxconnections = 15  # 最大连接数pool = PooledDB(pymysql,maxconnections,host='localhost',user='root',port=3306,passwd='123456',db='test_DB',use_unicode=True)return pool
# use
>> pool = mysql_connection()
>> con = pool.connection()

import pymysql
from DBUtils.PooledDB import PooledDB
import numpy as npclass MysqlPool():config = {'creator': pymysql,'host': 'localhost','port': 3306,'user': 'root','password': 'root','db': 'xiaogenggou','charset': 'utf8','maxconnections': 70,  # 连接池最大连接数量'cursorclass': pymysql.cursors.DictCursor}pool = PooledDB(**config)def __enter__(self):self.conn = MysqlPool.pool.connection()self.cursor = self.conn.cursor()return selfdef __exit__(self, type, value, trace):self.cursor.close()self.conn.close()# 执行函数
def execute(database_commands, s_value=None, i=0):""":param database_commands: 数据库的执行语句:param s_value: %s代表的变量名称:param i: 执行判断符，当i为1时执行状态判断 ,默认值为0"""# 错误捕获，并报错try:with MysqlPool() as db:# 执行操作命令 database_commands:数据库的查询语句 s_value:当查询语句出现%s时%s对应的值，默认为Nonedb.cursor.execute(database_commands, s_value)# 向数据库提交命令db.conn.commit()# 当执行查询操作时打印并返回查询的数据if "select" in database_commands and i == 0:# 打印查询的全部数据,fetchall()的返回值为list格式的字典集print(db.cursor.fetchall())# 返回查询的全部数据return db.cursor.fetchall()elif i == 1:return db.cursor.fetchall()except Exception as e:print("恭喜你，有错，又可以改bug了，开心吗")# 创建表的函数
def test_create():# 编写创建主表main_test的语句sql_create_test = "create table test" \"(user_id int auto_increment primary key," \"username varchar(11)," \"sex nvarchar(1)," \"age int(3)," \"state int(1) DEFAULT 1 )"# 创建表testexecute(sql_create_test)# test_create()# 删除表
def test_table_drop():# 编写删除表的语句sql_test_drop = "drop table test"# 删除表execute(sql_test_drop)# test_table_drop()# 向数据表内插入数据
def test_insert(username, sex, age, table="test"):""":param username: 用户名称:param sex: 用户性别:param age: 用户年龄:param table: 数据表的名称"""# 编写插入语句，%s表示元素的占位符sql_insert = "insert into " + table + " (username,sex,age) values(%s,%s,%s)"# 表示从左到右每个占位符所表示的元素的内容value = (username, sex, age)# 插入数据execute(sql_insert, value)'''
for i in range(10):value_username = "test" + str(i + 1)a = np.random.permutation(2)if a[1] == 1:value_sex = "男"else:value_sex = "女"value_age = int(20 + a[0])test_insert(value_username, value_sex, value_age)'''# 删除user_id对应的元组（用state值为0代替）
def test_delete(user_id):# 判断user_id对应的元组的状态if state_judgement(user_id) == 0:print("该行已被删除")else:# 编写删除语句sql_delete = "update test set state = %s where user_id = %s"# 将state的值改为0，代表删除value = (0, user_id)# 执行删除操作execute(sql_delete, value)'''
for i in range(1, 11):u_id = ia = np.random.permutation(2)print(a[1])if a[1] == 0:test_delete(u_id)
'''# 进行元组状态的判断，1表示存在，0表示已被删除
def state_judgement(user_id):""":param user_id: 判断状态的元组的主键(id):return: 状态判断值"""# 当user_id超出范围时报错try:# 编写判断的语句，%s表示元素的占位符sql_state_judgement = "select state from test where user_id = %s"# 表示从左到右每个占位符所表示的元素的内容value = user_id# 存储返回的listls = execute(sql_state_judgement, value, 1)# 返回代表状态的0或1return ls[0]["state"]except Exception as e:print("亲，错了呢")'''
for i in range(1, 11):print(state_judgement(i))
'''# 更改表test的数据
def test_update(event_attribute, new_value, user_id):""":param event_attribute: 要更改的属性名:param new_value: 更改后的属性内容:param user_id: 更改属性的元组的主键(id)"""# 判断user_id对应的元组的状态if state_judgement(user_id) == 0 and event_attribute != "state":print("该行已被删除")else:# 编写更新语句，%s表示元素的占位符sql_update = "update test set " + event_attribute + " = %s where user_id = %s "# 表示从左到右每个占位符所表示的元素的内容value = (new_value, user_id)# 更改数据execute(sql_update, value)'''
for i in range(1, 11):test_update("age", 40 + i, i)
'''# 查询表test中某一元组的所有的数据
def test_query_all(user_id=None):""":param user_id: 更改属性的元组的主键(id)"""# 当未指定user_id时查询全部元组的数据if user_id is None:# 编写查询语句sql_query_all = "select * from test where state = 1"# 执行查询execute(sql_query_all)else:# 判断user_id对应的元组的状态if state_judgement(user_id) == 0:print("该行已被删除")else:# 编写查询根据user_id确定的某一元组的全部数据sql_query_all = "select * from test where user_id = %s"# 表示从左到右每个占位符所表示的元素的内容value = user_id# 执行查询execute(sql_query_all, value)# test_query_all()# 查询副表test中某一元组的某一属性的数据
def test_query(event_attribute, user_id=None):""":param event_attribute:查询的属性的名称:param user_id: 更改属性的元组的主键(id)"""# 当未指定user_id时查询全部元组的数据if user_id is None:# 编写查询全部元组某一属性的语句sql_query = "select " + event_attribute + " from test where state = 1"# 执行查询execute(sql_query)else:# 判断user_id对应的元组的状态if state_judgement(user_id) == 0:print("该行已被删除")# 编写查询查询根据user_id确定的某一元组的某一属性的数据else:sql_query = "select " + event_attribute + " from test where user_id = %s"# 表示从左到右每个占位符所表示的元素的内容value = user_id# 执行查询execute(sql_query, value)# test_query("username")

总结

pymysql 和 mysqlclient 目前是python连接mysql 的主流方式。mysqlclient 速度比pymysql 更快；pymysql更加简单易使用。

python连接oracle

cx_Oracle

安装：cmd终端，执行 pip install cx_Oracle
依赖：oracle客户端，如果本地没有oracle客户的，就需要获取一个。Oracle Instant Client Downloads
解压后就是一个文件夹（instantclient_21_6），免安装。直接放到一个目录，例如：就直接放到C盘：C:\instantclient_21_6，然后把本地oracle客户端目录更新到环境变量；

import cx_Oracle #导入oracle包
connection =cx_Oracle.connect('aaaaa', 'bbb77', 'xx.xx.xx.xx:1521/SITE') # 数据库连接的用户名、密码、IP、端口和数据库名称。
cursor = connection.cursor() #数据库指针
cursor.execute("""SELECT first_name, last_name FROM employees 
WHERE department_id = :did AND employee_id > :eid""",did = 50,eid = 190)
#或者
sql="select a.* from hs_d.prce a where a.prod_code ='DB3045'" #定义要执行的sql语句
cursor.execute(sql) #执行sql语句
row = cursor.fetchall() #查询回来多条语句
print(row[0]) #打印查询的第一行结果 ---这里如果是测试的话，还可以assert。
print(row[0][2]) #打印结果是：DB3045
assert row[0][2]=="DB3045" #断言返回结果第一行的地3个字段值是“DB3045"

oracledb

import oracledb
import getpassuserpwd = getpass.getpass("Enter password: ")connection = oracledb.connect(user="hr", password=userpwd,dsn="dbhost.example.com/orclpdb")
# 其他功能代码# 关闭连接
connection.close()

或者可以使用以下方法

username="hr"
userpwd = os.environ.get("PYTHON_PASSWORD")
host = "localhost"
port = 1521
service_name = "orclpdb"dsn = f'{username}/{userpwd}@{port}:{host}/{service_name}'
connection = oracledb.connect(dsn)# 其他功能代码# 关闭连接
connection.close()

使用连接池的方法

# 初始化连接
pool = oracledb.create_pool(user="hr", password=userpwd, dsn="dbhost.example.com/orclpdb",min=2, max=5, increment=1)
# Acquire 连接到池
connection = pool.acquire()
# 使用连接池
with connection.cursor() as cursor:for result in cursor.execute("select * from mytab"):print(result)
# 释放连接池
pool.release(connection)
# 关闭连接池
pool.close()

python连接postgres

psycopg2

# -*- coding: utf-8 -*-
import psycopg2
# 获得连接
conn = psycopg2.connect(database="python", user="postgres", password="123456", host="127.0.0.1", port="5432")
# 获得游标对象，一个游标对象可以对数据库进行执行操作
cursor = conn.cursor()
# sql语句 建表
sql ="""SELECT * FROM student;"""
# 执行语句
cursor.execute(sql)
# 抓取
rows = cursor.fetchall()
print(rows)
# 事物提交
conn.commit()
# 关闭数据库连接
cursor.close()
conn.close()