使用SQLALchemy操作MySql数据库_sqlalchemy读取mysql

连接数据库并进行操作

[ORM概念]

是Object-Relational Mapping即对象关系映射，就是将关系型数据库和对象之间做一个映射，这样就可以不用使用SQL语句，直接用python语句就可以处理数据库了，其中，python中最成熟的ORM库就是SQLALchemy～

使用SQLALchemy进行数据库操作，需要三步，定义表（对应着数据库的表），与数据库进行连接，对数据库进行操作。

安装库

pip3 install sqlalchemy
# 如果pymysql驱动失败，需要安装
pip install mysql-connector
pip3 install mysql-connector-python
1
2
3
4

代码

# -*- coding:utf-8 -*-
# ---------------定义表需要的类-----------------
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, text, and_, or_, func
# --------------连接数据库需要的类---------------
from sqlalchemy import create_engine  # 建立数据库引擎
from sqlalchemy.orm import sessionmaker, relationship  # 建立会话session

# ---------第一部分：定义表-----------
from sqlalchemy.testing import in_

Base = declarative_base()  # 实例,创建基类


# 所有的表必须继承于Base
class Enhancer(Base):
    __tablename__ = 'my_test'  # 定义该表在mysql数据库中的实际名称
    # 定义表的内容
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    age = Column(Integer, nullable=False)
    gender = Column(Integer, nullable=False)


# ---------第二部分：与数据库连接--------
db_connect_string = 'mysql+mysqlconnector://root:123456@localhost:3306/my_db?charset=utf8'
# 以mysql数据库为例：mysql+数据库驱动：//用户名：密码@localhost:3306/数据库
engine = create_engine(db_connect_string)  # 创建引擎
Sesssion = sessionmaker(bind=engine)  # 产生会话
session = Sesssion()  # 创建Session实例
# ----------第三部分：进行数据操作--------
# 提交新数据
# session.add(Enhancer(name="测试", age=30, gender=1))  # 只能加一条数据
# session.add_all([Enhancer(name="Bob", age=29, gender=1), Enhancer(name="Huni", age=16, gender=1)])
# 使用add_all可以一次传入多条数据，以列表的形式。
# session.commit()  # 提交数据

# 获取数据
res = session.query(Enhancer).all()
print(res)
print(type(res))
for item in res:
    pass
    # print(item.name)

# 模糊查询like()
rs = session.query(Enhancer).filter(Enhancer.name.like('R%')).all()
# 通配符%, %ｃ代表以ｃ结尾，ｃ%代表以ｃ开头，%c%代表包含ｃ
for i in rs:
    print(i.id, i.name)

# 删除记录
session.query(Enhancer).filter(Enhancer.name == "测试").delete()
session.commit()

# 第一种更新记录方法
rs = session.query(Enhancer).filter(Enhancer.name.like('%i')).first()
# print(rs)  # 结果：<__main__.Enhancer object at 0x7fd7117d9a50>
# print(type(rs))  # 结果：<class '__main__.Enhancer'>
# rs.name = "Huya"
# 第二种更新记录方法
# synchronize_session=False,意思是不对session进行同步，直接更新数据库。
session.query(Enhancer).filter(Enhancer.name == "Huya").update({'name': 'Huy', 'age': 14}, synchronize_session=False)
session.commit()


# 多张表操作
# 定义两个表
class Class(Base):
    __tablename__ = 'my_class'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    lever = Column(Integer)
    address = Column(String(64))

    # 这里需要注意，并不是有一个studens列～，这里的students可以理解为
    # 关联的下一个表中满足条件的所有内容
    # backref用于反向查询
    students = relationship("Student", backref="my_class")


class Student(Base):
    __tablename__ = "student"
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    age = Column(Integer)
    class_id = Column(Integer, ForeignKey('my_class.id'))


# 班级表添加数据
# session.add_all([Class(name="一班", lever=1, address="二楼"), Class(name="二班", lever=2, address="三楼"),
#                  Class(name="三班", lever=3, address="八楼")])

# 学生表添加数据
# session.add_all([Student(name="Ryan", age=19, class_id=1), Student(name="Lucy", age=18, class_id=2),
#                  Student(name="Baby", age=19, class_id=3), Student(name="Huy", age=17, class_id=1),
#                  Student(name="Jack", age=22, class_id=2), Student(name="Niu", age=19, class_id=1)])
# session.commit()

# stu = session.query(Student).all()
# for i in stu:
#     print(i.name)

# 查找一班的所有学生
class_ = session.query(Class).filter(Class.name == "一班").first()
for i in class_.students:
    print(i.name)
print("---------------------------------------------------------------------")

# 补充查询
r1 = session.query(Student).all()
# filter传的是表达式，filter_by传的是参数
r3 = session.query(Student).filter(Student.name == "Ryan").all()
r4 = session.query(Student).filter_by(name='Ryan').all()
r5 = session.query(Student).filter_by(name='Ryan').first()
print(r5.age)
# :value 和:name 相当于占位符，用params传参数
r6 = session.query(Student).filter(text("id>:value and name=:name")).params(value=2, name='Ryan').order_by(
    Student.id).all()
print("r6:", r6)
# 自定义查询sql
r7 = session.query(Student).from_statement(text("SELECT * FROM student where name=:name")).params(name='Baby').first()
print("r7:", r7.name)

print("---------------------------------------------------------------------")

# 常用操作
# in_  在范围内
ret = session.query(Student).filter(Student.id.in_([1, 2, 3])).all()
print([i.name for i in ret])
# ~  非，除..之外
ret2 = session.query(Student).filter(~Student.id.in_([1, 2, 3])).all()
print([i.name for i in ret2])
# and_ 和 or_
ret3 = session.query(Student).filter(and_(Student.name == "Ryan", Student.age == 18)).all()
print([i.name for i in ret3])
ret4 = session.query(Student).filter(or_(Student.name == "Ryan", Student.age == 18)).all()
print([i.name for i in ret4])
# % 通配符,以R开头
ret5 = session.query(Student).filter(Student.name.like("R%")).all()
print([i.name for i in ret5])
# 排序，根据id降序排列（从大到小）
ret6 = session.query(Student).order_by(Student.id.desc()).all()
print([i.name for i in ret6])
# 第一个条件重复后，再按第二个条件升序排
ret7 = session.query(Student).order_by(Student.name.desc(), Student.id.asc()).all()
print([i.name for i in ret7])

print("---------------------------------------------------------------------")

# 分组
# 根据班级分组
ret8 = session.query(Class).group_by(Class.name).all()
print([i.name for i in ret8])
# 分组之后取最大id，id之和，最小id
ret9 = session.query(
    func.max(Class.id),
    func.sum(Class.id),
    func.min(Class.id)).group_by(Class.name).all()

# 连表（默认用forinkey关联）
res = session.query(Student, Class).filter(Student.class_id == Class.id).all()
for i in res:
    print(i.Student.name, i.Class.name)

# join表， 默认是inner join
res2 = session.query(Student).join(Class).all()
print([i.class_id for i in res2])
# isouter=True 外连，Student left join Class，没有右连接，反过来即可
res3 = session.query(Student).join(Class, isouter=True).all()
# 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
res4 = session.query(Student).join(Class, Student.class_id == Class.id, isouter=True).all()
for i in res4:
    print(i.name)
res5 = session.query(Student).join(Class, Student.class_id == Class.id, isouter=True)
print(res5)

# 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集
# union和union all的区别？
q1 = session.query(Student.name).filter(Student.id > 2)
q2 = session.query(Class.name).filter(Class.id < 2)
res6 = q1.union(q2).all()
print([i.name for i in res6])

q1 = session.query(Student.name).filter(Student.id > 2)
q2 = session.query(Class.name).filter(Class.id < 2)
res7 = q1.union_all(q2).all()
print([i.name for i in res7])

print("---------------------------------------------------------------------")

# 查询
cursor = session.execute('select * from my_class')
result = cursor.fetchall()
print(result)

# # 添加
# cursor = session.execute('insert into student(name,age,class_id) values(:value,:age,:class_id)',
#                          params={"value": 'lqz', "age": 18,
#                                  "class_id": 1})
# session.commit()
# print(cursor.lastrowid)

# 一对多查询
stu = session.query(Class).first()
print([i.name for i in stu.students])

session.close()

stu2 = session.query(Student, Class).join(Class, isouter=True).all()
for i in stu2:
    print(i[0].name, i[1].name)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212

多对多关系表

class Boy2Girl(Base):
    __tablename__ = 'boy2girl'
    id = Column(Integer, primary_key=True, autoincrement=True)
    girl_id = Column(Integer, ForeignKey('girl.id'))
    boy_id = Column(Integer, ForeignKey('boy.id'))


class Girl(Base):
    __tablename__ = 'girl'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)


class Boy(Base):
    __tablename__ = 'boy'

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
    
    # 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以
    servers = relationship('Girl', secondary='boy2girl', backref='boys')

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

常用操作

#　条件
ret = session.query(Users).filter_by(name='lqz').all()
#表达式，and条件连接
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
#注意下划线
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
#~非，除。。外
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
#二次筛选
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
#or_包裹的都是or条件，and_包裹的都是and条件
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符，以e开头，不以e开头
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制，用于分页，区间
ret = session.query(Users)[1:2]

# 排序，根据name降序排列（从大到小）
ret = session.query(Users).order_by(Users.name.desc()).all()
#第一个条件重复后，再按第二个条件升序排
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
#分组之后取最大id，id之和，最小id
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()
#haviing筛选
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表（默认用forinkey关联）

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
#join表，默认是inner join
ret = session.query(Person).join(Favor).all()
#isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可
ret = session.query(Person).join(Favor, isouter=True).all()
#打印原生sql
aa=session.query(Person).join(Favor, isouter=True)
print(aa)
# 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
ret = session.query(Person).join(Favor,Person.id==Favor.id, isouter=True).all()
# 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集
#union和union all的区别？
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

原生SQL

import time
import threading

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()

# 查询
# cursor = session.execute('select * from users')
# result = cursor.fetchall()

# 添加
cursor = session.execute('insert into users(name) values(:value)',params={"value":'lqz'})
session.commit()
print(cursor.lastrowid)

session.close()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

一对多关系

import time
import threading

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 添加
"""
session.add_all([
    Hobby(caption='乒乓球'),
    Hobby(caption='羽毛球'),
    Person(name='张三', hobby_id=3),
    Person(name='李四', hobby_id=4),
])

person = Person(name='张九', hobby=Hobby(caption='姑娘'))
session.add(person)
#添加二
hb = Hobby(caption='人妖')
hb.pers = [Person(name='文飞'), Person(name='博雅')]
session.add(hb)

session.commit()
"""

# 使用relationship正向查询
"""
v = session.query(Person).first()
print(v.name)
print(v.hobby.caption)
"""

# 使用relationship反向查询
"""
v = session.query(Hobby).first()
print(v.caption)
print(v.pers)
"""
#方式一，自己链表
# person_list=session.query(models.Person.name,models.Hobby.caption).join(models.Hobby,isouter=True).all()
person_list=session.query(models.Person,models.Hobby).join(models.Hobby,isouter=True).all()
for row in person_list:
    # print(row.name,row.caption)
    print(row[0].name,row[1].caption)

#方式二：通过relationship

person_list=session.query(models.Person).all()
for row in person_list:
    print(row.name,row.hobby.caption)
#查询喜欢姑娘的所有人
obj=session.query(models.Hobby).filter(models.Hobby.id==1).first()
persons=obj.pers
print(persons)
session.close()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

多对多关系

import time
import threading

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Group

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 添加
"""
session.add_all([
    Server(hostname='c1.com'),
    Server(hostname='c2.com'),
    Group(name='A组'),
    Group(name='B组'),
])
session.commit()

s2g = Server2Group(server_id=1, group_id=1)
session.add(s2g)
session.commit()


gp = Group(name='C组')
gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]
session.add(gp)
session.commit()


ser = Server(hostname='c6.com')
ser.groups = [Group(name='F组'),Group(name='G组')]
session.add(ser)
session.commit()
"""


# 使用relationship正向查询
"""
v = session.query(Group).first()
print(v.name)
print(v.servers)
"""

# 使用relationship反向查询
"""
v = session.query(Server).first()
print(v.hostname)
print(v.groups)
"""


session.close()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59