SQLAlchemy介绍及使用

发表于 更新于 分类于 ， Changyan： 本文字数： 1.6k 阅读时长 ≈ 6 分钟

ORM 全称 Object Relational Mapping, 翻译过来叫对象关系映射。简单的说，ORM 将数据库中的表与面向对象语言中的类建立了一种对应关系。这样，我们要操作数据库，数据库中的表或者表中的一条记录就可以直接通过操作类或者类实例来完成。

SQLAlchemy 是Python 社区最知名的 ORM 工具之一，为高效和高性能的数据库访问设计，实现了完整的企业级持久模型。

连接数据库&创建数据表

使用过django，使用其model很方便的帮我们管理数据，SQLAlchemy 负责的就是这个作用。

alchemy_db.py代码内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 导入基础模型
from sqlalchemy.ext.declarative import declarative_base
# 导入列、字段类型
from sqlalchemy import Column, Integer, String
# 导入数据库创建引擎
from sqlalchemy import create_conn
# 实例化模型基类
Base = declarative_base()

# 创建数据表对象
class User(Base):
    # 数据库中表真实名称
    __tablename__ = 'user'
    # 创建id列
    id = Column(Integer, comment="用户主键", primary_key=True, autoincrement=True)
    # 创建姓名列
    name = Column(String(32), comment='用户姓名')
# 创建数据库连接引擎
conn = create_conn("mysql+pymysql://root:@127.0.0.1:3306/sqlalchemysss?charset=utf8")
# 检索所有继承Base的模型对象，并在数据库中创建所有数据表
Base.metadata.create_all(conn)

增加数据

创建单条

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 导入会话
from sqlalchemy.orm import sessionmaker

# 导入数据库连接引擎和User模型
from alchemy_db import conn, User
print(User)

# 创建数据库会话
Session = sessionmaker(conn)
# 打开数据库会话
db_session = Session()

# 使用User模型创建一条数据
user_obj = User(name='张学友')
# 数据库会话中添加该数据
db_session.add(user_obj)


# 会话提交
db_session.commit()
# 关闭会话
db_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
# 导入会话
from sqlalchemy.orm import sessionmaker

# 导入数据库连接引擎和User模型
from alchemy_db import conn, User
print(User)

# 创建数据库会话
Session = sessionmaker(conn)
# 打开数据库会话
db_session = Session()


# 创建多条方法1
# 使用User模型创建一条数据
user_obj1 = User(name='刘德华')
# 使用User模型创建一条数据
user_obj2 = User(name='郭富城')
# 数据库会话中添加该数据
db_session.add(user_obj1)
# 数据库会话中添加该数据
db_session.add(user_obj2)

# 创建多条方法2
# 创建数据表模型列表
user_list = [
    User(name='孙悟空'),
    User(name='猪八戒')
]
# 提交到会话
db_session.add_all(user_list)


# 会话提交
db_session.commit()
# 关闭会话
db_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
# 导入数据库会话
from sqlalchemy.orm import sessionmaker

# 导入数据库连接引擎和数据模型
from alchemy_db import conn, User

# 创建会话
Session = sessionmaker(conn)
# 打开会话
db_session = Session()

# 查询所有数据
# 查询所有数据并按姓名正序
res = db_session.query(User).order_by(User.name.asc()).all()
# 循环读取查询到的数据
for item in res:
    print(item.id, '-', item.name)
    
# 只查询name列
res = db_session.query(User.name).all()
for i in res:
    print(i, i.name )

# 查询单条
res = sql.first()
print(res.id, res.name)

# 指定条件查询
res = db_session.query(User).filter(User.id == 3).first()
print(res.id, res.name)
res = db_session.query(User).filter_by(id = 3).first()
print(res.id, res.name)

# 关闭会话
db_session.close()

修改数据

1
2
3
4
5
6
7
8
9
10
11
12
13
from sqlalchemy.orm import sessionmaker

from alchemy_db import conn, User

Session = sessionmaker(conn)
db_session = Session()
# 更新一条
res = db_session.query(User).filter_by(id=5).update({"name":'布吉岛'})
# 更新多条
res = db_session.query(User).filter(User.id>5).update({"name":'小冤家'})
print(res)
db_session.commit()
db_session.close()

删除数据

1
2
3
4
5
6
7
8
9
10
11
12
13
from sqlalchemy.orm import sessionmaker

from alchemy_db import conn, User

Session = sessionmaker(conn)
db_session = Session()

# 删除一条
res = db_session.query(User).filter(User.id==7).delete()
print(res)

db_session.commit()
db_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
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
# 高级版查询操作,厉害了哦
#老规矩
from alchemy_db import User,conn
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(conn)
db_session = Session()

# 查询数据表操作
# and or
from sqlalchemy.sql import and_ , or_
ret = db_session.query(User).filter(and_(User.id > 3, User.name == '张学友')).all()
ret = db_session.query(User).filter(or_(User.id < 2, User.name == '张学友')).all()

# 查询所有数据
r1 = db_session.query(User).all()

# 查询数据 指定查询数据列 加入别名
r2 = db_session.query(User.name.label('username'), User.id).first()
print(r2.id,r2.username) # 15 张学友

# 表达式筛选条件
r3 = db_session.query(User).filter(User.name == "张学友").all()

# 原生SQL筛选条件
r4 = db_session.query(User).filter_by(name='张学友').all()
r5 = db_session.query(User).filter_by(name='张学友').first()

# 字符串匹配方式筛选条件 并使用 order_by进行排序
r6 = db_session.query(User).filter(text("id<:value and name=:name")).params(value=224, name='张学友').order_by(User.id).all()

#原生SQL查询
r7 = db_session.query(User).from_statement(text("SELECT * FROM User where name=:name")).params(name='张学友').all()

# 筛选查询列
# query的时候我们不在使用User ORM对象,而是使用User.name来对内容进行选取
user_list = db_session.query(User.name).all()
print(user_list)
for row in user_list:
    print(row.name)

# 别名映射  name as nick
user_list = db_session.query(User.name.label("nick")).all()
print(user_list)
for row in user_list:
    print(row.nick) # 这里要写别名了

# 筛选条件格式
user_list = db_session.query(User).filter(User.name == "张学友").all()
user_list = db_session.query(User).filter(User.name == "张学友").first()
user_list = db_session.query(User).filter_by(name="张学友").first()
for row in user_list:
    print(row.nick)

# 复杂查询
from sqlalchemy.sql import text
user_list = db_session.query(User).filter(text("id<:value and name=:name")).params(value=3,name="张学友")

# 查询语句
from sqlalchemy.sql import text
user_list = db_session.query(User).filter(text("select * from User id<:value and name=:name")).params(value=3,name="张学友")

# 排序 :
user_list = db_session.query(User).order_by(User.id).all()
user_list = db_session.query(User).order_by(User.id.desc()).all()
for row in user_list:
    print(row.name,row.id)

#其他查询条件
"""
ret = session.query(User).filter_by(name='张学友').all()
ret = session.query(User).filter(User.id > 1, User.name == '张学友').all()
ret = session.query(User).filter(User.id.between(1, 3), User.name == '张学友').all() # between 大于1小于3的
ret = session.query(User).filter(User.id.in_([1,3,4])).all() # in_([1,3,4]) 只查询id等于1,3,4的
ret = session.query(User).filter(~User.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查询不等于1,3,4的
ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='张学友'))).all() 子查询
from sqlalchemy import and_, or_
ret = session.query(User).filter(and_(User.id > 3, User.name == '张学友')).all()
ret = session.query(User).filter(or_(User.id < 2, User.name == '张学友')).all()
ret = session.query(User).filter(
    or_(
        User.id < 2,
        and_(User.name == 'eric', User.id > 3),
        User.extra != ""
    )).all()
# select * from User where id<2 or (name="eric" and id>3) or extra != "" 

# 通配符
ret = db_session.query(User).filter(User.name.like('e%')).all()
ret = db_session.query(User).filter(~User.name.like('e%')).all()

# 限制
ret = db_session.query(User)[1:2]

# 排序
ret = db_session.query(User).order_by(User.name.desc()).all()
ret = db_session.query(User).order_by(User.name.desc(), User.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = db_session.query(User).group_by(User.extra).all()
ret = db_session.query(
    func.max(User.id),
    func.sum(User.id),
    func.min(User.id)).group_by(User.name).all()

ret = db_session.query(
    func.max(User.id),
    func.sum(User.id),
    func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all()
"""

# 关闭连接
db_session.close()

高级修改

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#高级版更新操作
from alchemy_db import User,conn
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(conn)
db_session = Session()

#直接修改
db_session.query(User).filter(User.id > 0).update({"name" : "郑学友"})

#在原有值基础上添加 - 1
db_session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False)

#在原有值基础上添加 - 2
db_session.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate")
db_session.commit()