在python中操作mysql数据库，主要用到两个库，pymysql和 sqlalchemy。分别进行介绍
安装
安装没啥好说的，其实就是pip install就完事
pip install pymysql pip install sqlalchemy
pymsql操作数据库
创建连接
以下语句省略了import语句，请自行import
要操作数据库，首先需要创建和数据库的连接，然后才能进程CRUD的操作。
# pymysql用connect方法进行连接
conn = pymysql.connect(host="192.168.32.11", port=3306,
user="hellokitty", password="123123",
database="hrs", charset="utf8mb4")
插入数据Create
no = int(input('部门编号: '))
name = input('部门名称: ')
location = input('部门所在地: ')
try:
# 2.获取游标对象
with conn.cursor() as cursor:
# 3.通过游标对象对数据库服务器发出sql语句
affected_rows = cursor.execute(
f"insert into `tb_dept` values (%s,%s,%s)",
(no, name, location)
)
if affected_rows == 1:
print("新增部门成功")
# 4.提交
conn.commit()
except pymysql.MySQLError as err:
# 4.回滚
conn.rollback()
print(type(err), err)
finally:
# 5.关闭连接
conn.close()
插入多组可以参考如下：
with conn.cursor() as cursor:
affected_rows = cursor.executemany(
'insert into `tb_dept` values (%s, %s, %s)',
[(no, name, location),
(no1, name1, location1),
(no2, name2, location2)]
)
if affected_rows == 3:
print('新增部门成功!!!')
conn.commit()
删除delete
该例子中指定了一个参数：autocommit=True，这样SQL代码就会自动提交。实际环境中不建议这样做。
no = int(input("请输入部门编号："))
conn = pymysql.connect(host="192.168.32.11", port=3306,
user="hellokitty", password="123123",
database="hrs", charset="utf8mb4",
autocommit=True)
try:
with conn.cursor() as cursor:
affected_rows = cursor.execute(
"delete from `tb_dept` where `dno`=%s",
(no,)
)
if affected_rows == 1:
print("删除部门成功")
finally:
conn.close()
更新update
no = int(input('部门编号: '))
name = input('部门名称: ')
location = input('部门所在地: ')
try:
with conn.cursor() as cursor:
affected_rows = cursor.execute(
"update `tb_dept` set `dname`=%s, `dloc`=%s where `dno`=%s",
(name, location, no)
)
if affected_rows == 1:
print("更新部门信息成功")
conn.commit()
except pymysql.MySQLError as err:
conn.rollback()
print(type(err), err)
finally:
conn.close()
查询数据
try:
with conn.cursor() as cursor:
affected_rows = cursor.execute(
"select `dno`,`dname`,`dloc` from `tb_dept`")
row = cursor.fetchone()
while row:
print(row)
row = cursor.fetchone()
except pymysql.MySQLError as err:
conn.rollback()
print(type(err), err)
finally:
conn.close()
本例子中，使用while循环每次用fetchone获取一条数据然后打印。也提供了fetchall方法可以获取到所有的结果，但是不推荐这样做，因为在实际环境中这样对内存的压力很大。
默认拿到的是元组，如果希望拿到列表，需要在连接数据库时指定cursorclass=DictCursor
pymysql的缺陷
通过以上例子应该可以看出来，实际上pymysql操作数据库就是通过执行sql语句来的。通过拼接字符串的方式写好sql语句，然后交给pymysql执行。这种方式的缺点是显而易见的：
- sql语句比较长的时候很不方便
- 可能会有sql注入的危险
- 不太优雅
为了解决这些问题，可以使用sqlalchemy库
使用sqlalchemy库操作数据库
sqlalchemy创建数据库连接
首先用create_engine方法创建数据库引擎，然后再用connect方法连接。这里要稍微注意以下，第一个参数数据库的URI这里，不像pymysql直接输入就可以了，比如说pwd_alchemy = 'abc%401234’，这里实际密码是abc@1234，但不能直接输入@，需转换成%40。
from sqlalchemy import create_engine
# 创建连接
engine = create_engine(
DB_URI, # "mysql+pymysql://{USERNAME}:{pwd_alchemy}@{HOST}:{PORT}/{DATABASE}?charset=utf8mb4",
echo=False, # echo 设为 True 会打印出实际执行的 sql，调试的时候更方便
future=True, # 使用 SQLAlchemy 2.0 API，向后兼容
pool_size=5, # 连接池的大小默认为 5 个，设置为 0 时表示连接无限制
pool_recycle=3600, # 设置时间以限制数据库自动断开
)
with engine.connect() as conn:
......
执行sql语句
sqlalchemy也可以通过执行sql语句的方式操作数据库，这部分和pymysql区别不大。但是execute方法执行的sql语句，需要用sqlalchemy的text方法进行封装，这一点需要注意。
from sqlalchemy import text
sql_text = "select * from tb_keys"
with engine.connect() as conn:
result = conn.execute(text(sql_text))
# 查询结果result类似生成器, 只能遍历一遍, 遍历第二遍时就是空数据
# print(result.all())
res = result.all()
# result可以遍历，每一行是一个row对象,类似具名元祖(namedtuple)，支持以下2种遍历方式
for row in res:
print(row.keys_id, row.keys_name, row.keys_count) # 通过字段名获取
# print(row[0], row[1], row[2]) # 通过索引获取
顺便执行多条的语句，也很类似
# with engine.connect() as conn:
# data = [{"keys_id": 11, "keys_name": 'test1', "keys_count": 1},
# {"keys_id": 12, "keys_name": 'test2', "keys_count": 1}]
# conn.execute(
# text("INSERT INTO tb_keys (keys_id, keys_name, keys_count) VALUES (:keys_id, :keys_name, :keys_count)"),
# data
# )
# # 手动commit
# conn.commit()
声明式API
接下来重点说明一下sqlalchemy的声明式API。这就相当于直接创建一个Table对象，如下所示。
from sqlalchemy.orm import DeclarativeBase, Session
class Base(DeclarativeBase):
"""DeclarativeBase无法直接使用，所以要先继承一个Base类"""
pass
class TableCount(Base):
__tablename__ = "tb_keys"
keys_id: Mapped[int] = mapped_column(Integer, primary_key=True)
keys_name: Mapped[str] = mapped_column(String(30), index=True)
keys_count: Mapped[int] = mapped_column(Integer)
我这张表结果很简单，表名是tb_keys，然后id，name，count3个字段，分别是主键、字符串和int类型。
用声明式api进行增删改查
然后是增删改查的例子：
from sqlalchemy import select, update
# 用声明式API进行select查找，而不是直接执行sql语句
with Session(engine) as session:
stmt = select(TableCount).where(TableCount.keys_count == 1).order_by(TableCount.keys_id)
result = session.execute(stmt)
# 一般情况下，当选取整个对象的时候，都要用 scalars 方法
res2 = result.scalars()
for row in res2:
print(row.keys_id, row.keys_name, row.keys_count)
print("*" * 40)
# 查询单个属性，不需要用
res3 = session.execute(select(TableCount.keys_name))
for row in res3:
print(row.keys_name)
print("*" * 40)
# 查询主键有一个快捷方式，以下查询id是7
key_word = session.get(TableCount, 7)
print(key_word.keys_name)
# 更新数据使用update
stmt = update(TableCount).where(TableCount.keys_name == "护士").values(keys_name="护师").\
execution_options(synchronize_session="fetch")
session.execute(stmt)
# 也可以直接修改值，比如上面获取到的
key_word.keys_name = "Nurse"
session.commit()
# 注意，以下两种方式都能更新count值，但更推荐第二种做法。第一种方式可能会导致竞争更新，race condition（竞态条件
# key_word.keys_count += 1
key_word.keys_count = TableCount.keys_count + 1
session.commit()
# 新增
# new_word = TableCount()
# new_word.keys_count = 0
# new_word.keys_name = "Alice"
# session.add(new_word)
# session.commit()
# 删除， 用session.delete 删除，先获取到id，在get到该对象，然后用session.delete删除
del_word = session.execute(select(TableCount.keys_id).where(TableCount.keys_name == 'Alice')).fetchone()
del_word_id = del_word[0]
del_word_obj = session.get(TableCount, del_word_id)
session.delete(del_word_obj)
session.commit()
完整代码请参考：https://github.com/h-kayotin/hanayo_hr/blob/master/hanayo_hr/db_sqlalchemy.py
![[工业自动化-1]：PLC架构与工作原理](https://img-blog.csdnimg.cn/ce10a1471ed14382bc58364cf8bd5209.png)
还没有评论，来说两句吧...