Python3 之 MySQL操作模块：pymsql和sqlachemy

Posted 2020-10-06

Python操作MySQL主要使用两种方式：

• 原生模块 pymsql
• ORM框架 sqlachemy

pymsql

下载安装

pip3 install pymysql

使用操作

1、执行SQL

 1 # -*- coding:utf-8 -*-
 2 
 3 import pymysql
 4   
 5 # 创建连接
 6 conn = pymysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘t1‘)
 7 # 创建游标
 8 cursor = conn.cursor()
 9   
10 # 执行SQL，并返回收影响行数
11 effect_row = cursor.execute("update hosts set host = ‘1.1.1.2‘")
12   
13 # 执行SQL，并返回受影响行数
14 #effect_row = cursor.execute("update hosts set host = ‘1.1.1.2‘ where nid > %s", (1,))
15   
16 # 执行SQL，并返回受影响行数
17 #effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
18   
19   
20 # 提交，不然无法保存新建或者修改的数据
21 conn.commit()
22   
23 # 关闭游标
24 cursor.close()
25 # 关闭连接
26 conn.close()

2、获取新创建数据自增ID

 1 # -*- coding:utf-8 -*-
 2 import pymysql
 3   
 4 conn = pymysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘t1‘)
 5 cursor = conn.cursor()
 6 cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
 7 conn.commit()
 8 cursor.close()
 9 conn.close()
10   
11 # 获取最新自增ID
12 new_id = cursor.lastrowid

3、获取查询数据

 1 # -*- coding:utf-8 -*-
 2 import pymysql
 3   
 4 conn = pymysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘t1‘)
 5 cursor = conn.cursor()
 6 cursor.execute("select * from hosts")
 7   
 8 # 获取第一行数据
 9 row_1 = cursor.fetchone()
10   
11 # 获取前n行数据
12 # row_2 = cursor.fetchmany(3)
13 # 获取所有数据
14 # row_3 = cursor.fetchall()
15   
16 conn.commit()
17 cursor.close()
18 conn.close()

注：在fetch数据时按照顺序进行，可以使用cursor.scroll(num,mode)来移动游标位置，如：

• cursor.scroll(1,mode=‘relative‘)  # 相对当前位置移动
• cursor.scroll(2,mode=‘absolute‘) # 相对绝对位置移动

4、fetch数据类型

　　关于默认获取的数据是元祖类型，如果想要或者字典类型的数据，即：

 1 # -*- coding:utf-8 -*-
 2 import pymysql
 3   
 4 conn = pymysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘t1‘)
 5   
 6 # 游标设置为字典类型
 7 cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
 8 r = cursor.execute("call p1()")
 9   
10 result = cursor.fetchone()
11   
12 conn.commit()
13 cursor.close()
14 conn.close()

 

sqlachemy

sqlachemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

安装：

pip3 install sqlachemy

sqlachemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如以pymsql为例：

mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]

一、内部处理

使用 Engine/ConnectionPooling/Dialect 进行数据库操作，Engine使用ConnectionPooling连接数据库，然后再通过Dialect执行SQL语句。

 1 # -*- coding:utf-8 -*-
 2 from sqlalchemy import create_engine
 3   
 4   
 5 engine = create_engine("mysql+pymysql://root:[email protected]:3306/t1", max_overflow=5)
 6   
 7 # 执行SQL
 8 # cur = engine.execute(
 9 #     "INSERT INTO hosts (host, color_id) VALUES (‘1.1.1.22‘, 3)"
10 # )
11   
12 # 新插入行自增ID
13 # cur.lastrowid
14   
15 # 执行SQL
16 # cur = engine.execute(
17 #     "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[(‘1.1.1.22‘, 3),(‘1.1.1.221‘, 3),]
18 # )
19   
20   
21 # 执行SQL
22 # cur = engine.execute(
23 #     "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)",
24 #     host=‘1.1.1.99‘, color_id=3
25 # )
26   
27 # 执行SQL
28 # cur = engine.execute(‘select * from hosts‘)
29 # 获取第一行数据
30 # cur.fetchone()
31 # 获取第n行数据
32 # cur.fetchmany(3)
33 # 获取所有数据
34 # cur.fetchall()

二、ORM功能使用

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有组件对数据进行操作。根据类创建对象，对象转换成SQL，执行SQL。

1、创建表

表结构 + 数据库连接

 1 # -*- coding:utf-8 -*-
 2 from sqlalchemy.ext.declarative import declarative_base
 3 from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
 4 from sqlalchemy.orm import sessionmaker, relationship
 5 from sqlalchemy import create_engine
 6 
 7 engine = create_engine("mysql+pymysql://root:[email protected]:3306/t1", max_overflow=5)
 8 
 9 Base = declarative_base()
10 
11 # 创建单表
12 class Users(Base):
13     __tablename__ = ‘users‘
14     id = Column(Integer, primary_key=True)
15     name = Column(String(32))
16     extra = Column(String(16))
17 
18     __table_args__ = (
19     UniqueConstraint(‘id‘, ‘name‘, name=‘uix_id_name‘),
20         Index(‘ix_id_name‘, ‘name‘, ‘extra‘),
21     )
22 
23     def __repr__(self):
24         return "%s-%s" %(self.id, self.name)
25 
26 # 一对多
27 class Favor(Base):
28     __tablename__ = ‘favor‘
29     nid = Column(Integer, primary_key=True)
30     caption = Column(String(50), default=‘red‘, unique=True)
31 
32     def __repr__(self):
33         return "%s-%s" %(self.nid, self.caption)
34 
35 class Person(Base):
36     __tablename__ = ‘person‘
37     nid = Column(Integer, primary_key=True)
38     name = Column(String(32), index=True, nullable=True)
39     favor_id = Column(Integer, ForeignKey("favor.nid"))
40     # 与生成表结构无关，仅用于查询方便
41     favor = relationship("Favor", backref=‘pers‘)
42 
43 # 多对多
44 class ServerToGroup(Base):
45     __tablename__ = ‘servertogroup‘
46     nid = Column(Integer, primary_key=True, autoincrement=True)
47     server_id = Column(Integer, ForeignKey(‘server.id‘))
48     group_id = Column(Integer, ForeignKey(‘group.id‘))
49     group = relationship("Group", backref=‘s2g‘)
50     server = relationship("Server", backref=‘s2g‘)
51 
52 class Group(Base):
53     __tablename__ = ‘group‘
54     id = Column(Integer, primary_key=True)
55     name = Column(String(64), unique=True, nullable=False)
56     port = Column(Integer, default=22)
57     # group = relationship(‘Group‘,secondary=ServerToGroup,backref=‘host_list‘)
58 
59 
60 class Server(Base):
61     __tablename__ = ‘server‘
62 
63     id = Column(Integer, primary_key=True, autoincrement=True)
64     hostname = Column(String(64), unique=True, nullable=False)
65 
66 
67 
68 
69 def init_db():
70     Base.metadata.create_all(engine)
71 
72 
73 def drop_db():
74     Base.metadata.drop_all(engine)
75 
76 
77 Session = sessionmaker(bind=engine)
78 session = Session()

2、操作表

 1 # -*- coding:utf-8 -*-
 2  
 3 from sqlalchemy.ext.declarative import declarative_base
 4 from sqlalchemy import Column, Integer, String
 5 from sqlalchemy.orm import sessionmaker
 6 from sqlalchemy import create_engine
 7  
 8 engine = create_engine("mysql+mysqldb://root:[email protected]:3306/s11", max_overflow=5)
 9  
10 Base = declarative_base()
11  
12  
13 class User(Base):
14     __tablename__ = ‘users‘
15     id = Column(Integer, primary_key=True)
16     name = Column(String(50))
17  
18 # 寻找Base的所有子类，按照子类的结构在数据库中生成对应的数据表信息
19 # Base.metadata.create_all(engine)
20  
21 Session = sessionmaker(bind=engine)
22 session = Session()
23  
24  
25 # ########## 增 ##########
26 # u = User(id=2, name=‘sb‘)
27 # session.add(u)
28 # session.add_all([
29 #     User(id=3, name=‘sb‘),
30 #     User(id=4, name=‘sb‘)
31 # ])
32 # session.commit()
33  
34 # ########## 删除 ##########
35 # session.query(User).filter(User.id > 2).delete()
36 # session.commit()
37  
38 # ########## 修改 ##########
39 # session.query(User).filter(User.id > 2).update({‘cluster_id‘ : 0})
40 # session.commit()
41 # ########## 查 ##########
42 # ret = session.query(User).filter_by(name=‘sb‘).first()
43  
44 # ret = session.query(User).filter_by(name=‘sb‘).all()
45 # print ret
46  
47 # ret = session.query(User).filter(User.name.in_([‘sb‘,‘bb‘])).all()
48 # print ret
49  
50 # ret = session.query(User.name.label(‘name_label‘)).all()
51 # print ret,type(ret)
52  
53 # ret = session.query(User).order_by(User.id).all()
54 # print ret
55  
56 # ret = session.query(User).order_by(User.id)[1:3]
57 # print ret
58 # session.commit()

 

# -*- coding:utf-8 -*-

 

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column, Integer, String

from sqlalchemy.orm import sessionmaker

from sqlalchemy import create_engine

 

engine = create_engine("mysql+mysqldb://root:[email protected]:3306/s11", max_overflow=5)

 

Base = declarative_base()

 

 

class User(Base):

    __tablename__ = ‘users‘

    id = Column(Integer, primary_key=True)

    name = Column(String(50))

 

# 寻找Base的所有子类，按照子类的结构在数据库中生成对应的数据表信息

# Base.metadata.create_all(engine)

 

Session = sessionmaker(bind=engine)

session = Session()

 

 

# ########## 增 ##########

# u = User(id=2, name=‘sb‘)

# session.add(u)

# session.add_all([

#     User(id=3, name=‘sb‘),

#     User(id=4, name=‘sb‘)

# ])

# session.commit()

 

# ########## 删除 ##########

# session.query(User).filter(User.id > 2).delete()

# session.commit()

 

# ########## 修改 ##########

# session.query(User).filter(User.id > 2).update({‘cluster_id‘ : 0})

# session.commit()

# ########## 查 ##########

# ret = session.query(User).filter_by(name=‘sb‘).first()

 

# ret = session.query(User).filter_by(name=‘sb‘).all()

# print ret

 

# ret = session.query(User).filter(User.name.in_([‘sb‘,‘bb‘])).all()

# print ret

 

# ret = session.query(User.name.label(‘name_label‘)).all()

# print ret,type(ret)

 

# ret = session.query(User).order_by(User.id).all()

# print ret

 

# ret = session.query(User).order_by(User.id)[1:3]

# print ret

# session.commit()