Python精进爆肝几天，全网最全的SQLAlchemy框架使用手册，值得一看

Posted 2021-08-16 码农飞哥

您好，我是码农飞哥，感谢您阅读本文！如果此文对您有所帮助，请毫不犹豫的一键三连吧。小伙伴们，有啥想看的，想问的，欢迎积极留言告诉我喔。 事情是这样的，做测试的女朋友这段时间正在学习Python。这天晚上我下班回家，看到她一个人正在学习SQLAlchemy。真是个好学的girl，这时候我意识到自己表现的机会来了。这个我懂呀。然后，就有了下面这篇文章。

文章目录

• • SQLAlchemy是什么
• 准备工作(没有时间的小伙伴可以直接跳到实战章节)
• • 0. 数据表准备
  • 1. 添加数据库配置
  • 3. 创建数据表对应的model
• 实战开发（重点）
• • • 新增数据
  • 牛刀小试（查询操作）
  • • 1. 根据主键查询数据
    • 2. 多条件查询部分字段
    • 3. or查询
    • 4. like查询
    • 5. 分页查询
    • 6.关联查询
    • 7. 判断数据是否存在
    • ColumnOperators类介绍
    • 常用表达式的介绍
    • 更新数据
    • • 第一种更新数据的方式
      • 第二种更新数据的方式
  • 删除数据
  • 推荐阅读
  • 总结

SQLAlchemy是什么

SQLAlchemy是一种应用于Python语言中的ORM框架。 ORM的全称是Object-Relational Mapping，就是把关系数据库的表结构映射到对象上。

准备工作(没有时间的小伙伴可以直接跳到实战章节)

俗话说的好，兵马未动，粮草先行，要想学习ORM框架我们是不是得先创建两张测试表。我啪的一下拿出来之前我创建的两张数据表，一张是用户表，一张是博客表。场景就是围绕着用户发布博客，查询博客展开。

0. 数据表准备

-- 用户表
CREATE TABLE `user` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `username` varchar(35) NOT NULL COMMENT '用户名',
  `password` varchar(35) NOT NULL COMMENT '密码',
  PRIMARY KEY (`id`),
  UNIQUE KEY `username` (`username`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COMMENT='用户表';
-- 文章表
CREATE TABLE `post` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `author_id` int(11) NOT NULL COMMENT '作者id',
  `title` varchar(50) NOT NULL COMMENT '标题',
  `body` varchar(2048) NOT NULL COMMENT '正文',
  `created` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='文章表';

1. 添加数据库配置

数据表弄好之后就是创建一个初始化文件init.py，路径是 flaskr/config.py，这里主要添加的是数据库配置，如果后期有其他配置也可以在该文件里面。

# 初始化数据库连接
engine = create_engine('mysql+mysqlconnector://root:123@localhost:3306/job')
# 创建DBSession类型
DBSession = sessionmaker(bind=engine)
#创建session对象
session = DBSession()

3. 创建数据表对应的model

SQLAlchemy是一个ORM框架，就是将数据表与对象进行对应的。所以，针对前面的user表和post表，我们分别建立interactive_user和interactive_post两个model与之对应。下面以user表为例，文件地址：flaskr/model/interactive_user.py

from flaskr import db
#定义user表对应的model类InteractiveUser
class InteractiveUser(db.Model):
    __tablename__ = 'user'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    password = db.Column(db.String(80), unique=True)

    # 序列化
    @property
    def serialize(self):
        return {
            'id': self.id,
            'username': self.username,
            'password': self.password
        }

---

实战开发（重点）

前面的准备工作做完之后，女票嘴角微微一笑，接下来就重头戏了，学学SQLAlchemy框架的使用。这一章主要介绍其增删改查，分页查询，关联查询等各种查询的使用。

新增数据

女票说，我们先来编写一个新增方法，插入测试数据吧。在我们一番探讨下写出了下面的方法。

def insert_post(title, body, author_id):
    try:
        post = InteractivePost()
        post.title = title
        post.body = body
        post.author_id = author_id
        session.add(post)
        session.flush()
        return post
    except SQLAlchemyError as e:
        session.rollback()
        raise e

由于需要插入的字段比较少，所以，这里将每个字段作为一个参数进行了传入。如果参数比较多的话，则可以直接传入一个对象。比如：传入InteractivePost对象，用户表的新增也是同理。

牛刀小试（查询操作）

我们插入了好几条测试数据，接下来自然而然的就是编写查询语句。

1. 根据主键查询数据

首先，让我们小试一下牛刀，如下，通过session对象调用query方法获取查询的字段，如果是查询表中所有字段就直接传入表对应的model，如本例中的InteractiveUser。get方法直接返回了查询后的结果，因为是主键查询，所以，返回的数据只有一条，可以直接用InteractiveUser的对象来接收。

user = session.query(InteractiveUser).get(id)

session.query(InteractiveUser) 返回的是BaseQuery对象，对应的sql语句是：

SELECT user.id AS user_id, user.username AS user_username, user.password AS user_password 
FROM user

如下图所示：BaseQuery对象有很多内容，包括_distinct等关键词。

2. 多条件查询部分字段

做完了简单的查询，女票又说了，该如何筛选某个作者发布的某一篇文章呢？这时候就不能用get方法了，因为get方法只能用于主键查询，其只能返回整个表的数据。所以要换成filter方法了，这个方法你可以当成是SQL语句中的where关键词，所有的查询条件都通过它来传入。

post_tuple = session.query(InteractivePost.title, InteractivePost.body).filter(InteractivePost.title == title,                                                                             InteractivePost.author_id==author_id).first()

这个语句执行对应的SQL是：

SELECT post.title AS post_title, post.body AS post_body 
FROM post 
WHERE post.title = %(title_1)s AND post.author_id = %(author_id_1)s

query方法是一个有可变参数的方法，所以可以传入我们需要查询的字段。这里传入了title和body两个字段。
filter方法传入的是查询条件，这里传入了title和author_id两个查询条件，默认是and查询。并且都是精确查询
first()方法表示返回所有满足条件的记录的第一条。如果要返回所有满足条件的记录，则需要使用all()方法。
需要注意的是这里返回的数据是一个tuple类型的数据。

3. or查询

数据完美的查询出来之后女票又说了该如何实现or查询呢？这个也很简单，只需要引入or_方法。然后在filter方法中传入or_方法，把需要通过or查询的条件传入到or_方法中。

from sqlalchemy import or_
post_tuple = session.query(InteractivePost.title, InteractivePost.body).filter(
        or_(InteractivePost.title == title,
            InteractivePost.author_id)).all()
    return post_tuple

这个方法等价的SQL是，可以看到post.title和post.author_id两个查询条件通过or相关联。

SELECT post.title AS post_title, post.body AS post_body 
FROM post 
WHERE post.title = %(title_1)s OR post.author_id = %(author_id_1)s

4. like查询

or查询搞完了，但是新的问题又来了，该如何根据文章标题来模糊查询文章呢？实际应用中基本上不会对标题做精确查询。

    post_list = session.query(InteractivePost).filter(InteractivePost.title.like('%{0}%'.format(title))).all()
    return [post.serialize for post in post_list]

这个也很简单啦，只需要在模糊查询字段上调用like方法，是全匹配查询。这里的like方法ColumnOperators类中的方法，后面会详细介绍这个类。
这个方法等价的SQL是：

SELECT post.id AS post_id, post.author_id AS post_author_id, post.created AS post_created, post.title AS post_title, post.body AS post_body 
FROM post 
WHERE post.title LIKE %(title_1)s

5. 分页查询

一些基本的查询弄好了，但是列表的分页查询我们还没有弄呢。那么问题又来了，如何根据title和author_id来分页查询文章呢？并且title和author_id都是非必传的

# 分页查询文章
def page_post(title, author_id, current_page, page_size):
    session = session.query(InteractivePost)
    # 判断title是否为空，不为空的传入该条件
    if title:
        session = session.filter(InteractivePost.title == title)
    if author_id:
        session = session.filter(InteractivePost.author_id == author_id)
    # 统计条数
    total = session.count()
    # 分页查询
    post_list = session.order_by(InteractivePost.created.desc()) \\
        .limit(int(page_size)) \\
        .offset((int(current_page) - 1) * int(page_size)).all()
    if post_list is not None:
        return total, [post.serialize for post in post_list]
    else:
        return total, None

首先是创建好session对象，如果title不为空，则会传入title条件。同样的如果author_id不为空的话，则传入author_id条件。
调用count()方法会返回满足条件的总记录数。
调用order_by方法进行排序操作，在本例中是按照创建时间created进行降序。
limit 方法是返回的记录数。
offset 方法是索引的偏移量。
这个分页查询其实有两条SQL语句，一条是统计总记录数的SQL语句，一个是分页查询文章记录的SQL语句。

SELECT post.id AS post_id, post.author_id AS post_author_id, post.created AS post_created, post.title AS post_title, post.body AS post_body 
FROM post 
WHERE post.title = %(title_1)s 
 ORDER BY post.created DESC 
 LIMIT %(param_1)s, %(param_2)s

6.关联查询

做完了分页查询之后，女票又提出了一个问题，文章表中没有作者名称了，该怎么在查询文章的同时关联带出文章的作者名称呢？这个当然也很好处理的啦！

# 关联查询用户数据
def post_join_author(title):
    title_list = session \\
        .query(InteractivePost.id, InteractivePost.title,
               InteractivePost.body, InteractiveUser.username) \\
        .join(InteractiveUser, InteractivePost.author_id == InteractiveUser.id) \\
        .filter(InteractivePost.title == title).all()
    if title_list is not None:
        return [{'id': title[0],
                 'title': title[1],
                 'body': title[2],
                 'username': title[3]} for title in title_list]
    else:
        return None

这里通过join方法来进行关联查询，这里是内连接通过author_id字段进行关联的。
上面语句执行的SQL语句是：

SELECT post.id AS post_id, post.title AS post_title, post.body AS post_body, user.username AS user_username 
FROM post INNER JOIN user ON post.author_id = user.id 
WHERE post.title = %(title_1)s

可以看到这里的查询是内连接，如果是左连接的需要用outerjoin方法。
查询返回结果是：

7. 判断数据是否存在

在首页根据登陆时，如何根据用户名和密码查询该用户是否存在呢？

def is_exist_user(username, password):
    result = session.query(exists().where(and_(InteractiveUser.username == username,
                                                  InteractiveUser.password == password))).scalar()
    return result

上面语句对应的SQL语句如下：

SELECT EXISTS (SELECT * 
FROM user 
WHERE user.username = %(username_1)s AND user.password = %(password_1)s) AS anon_1

根据用户名和密码查询用户是否存在，这里的result是bool类型，关键的方法是scalar方法，通过where方法传入查询条件，多个条件的话需要通过and_方法包装。

ColumnOperators类介绍

ColumnOperators类是是一个非常重要的类，由名字可以知道它主要的方法都是列操作相关的方法，比如列的排序，like查询等

方法	对应的sql语句的关键词	作用
like	like	用于条件的模糊查询
in_	in	用于in查询
union	union	用于关联两个查询结果
desc	desc	用于对返回结果进行降序
asc	asc	用于对返回结果进行升序
distinct	distinct	用于对字段进行去重
startswith	startswith	查询以特定开头内容的数据，比如： InteractivePost.title.startswith(“测试”) 就是查询标题以 " 测试" 开头文章，相当于左匹配
endswith	endswith	查询以特定结尾内容的数据，比如： InteractivePost.title.endswith(“测试”) 就是查询标题以 " 测试" 结尾文章，相当于右匹配
contains	contains	查询包含特定内容的数据，比如： InteractivePost.title.contains(“测试”) 就是查询标题包含 " 测试" 二字的文章，相当于全匹配

常用表达式的介绍

想上面提到的and_，or_等表达式都是比较常用的表达式，在此也一并介绍一下。

表达式	对应的sql语句的关键词	作用
and_	and	用于查询条件之间并的关系
or_	or	用于查询条件之间或的关系
union	union	用于关联两个不同的查询，会去重
union	union all	用于关联多个不同的查询，不会去重
union_all	union all	用于关联多个不同的查询，不会去重
distinct	distinct	去重，与前面ColumnOperators中的distinct不同的是，前面的是针对单个字段的
exists	exists	判断数据是否存在
select	select	查询数据的DML语句
insert	insert	插入数据的DML语句
update	update	更新数据的DML语句
delete	delete	删除数据的DML语句
join	inner join	两个表之间内连接的语句
outerjoin	outer join	两个表之间外连接的语句，有左外连接和右外连接

更新数据

更新数据的方式有两种，第一种是直接查出待更新的数据，然后重新set新值。

第一种更新数据的方式

def update_model_state(id, username,password):
    user = session.query(InteractiveUser).filter(InteractiveUser.id == id).first()
    user.password = password
	user.username=username
    session.commit()
    return user

第二种方式是直接调用update方法，需要注意的是update方法中传入的是一个dict类型（键值对）的数据。

第二种更新数据的方式

def update_model_state2(id, username, password):
    session.query(InteractiveUser).filter(InteractiveUser.id == id).update(
        {'password': password, 'username': username})
    session.commit()

删除数据

删除数据跟更新差不多，其实就是多调用了一个session.delete(post)方法。

def del_post(id):
    post = session.query(InteractivePost).filter(InteractivePost.id == id).first()
    if post is None:
        return None
    else:
        session.delete(post)
        session.commit()
    return post

推荐阅读

【两分钟了解Python之SQLAlchemy框架的使用】

【Python中如何编写接口，以及如何请求外部接口】

【一分钟快速实现Flask框架与SQLAlchemy框架的整合】

总结

本文详细介绍了SQLAlchemy框架的使用。以对话的方式来叙述文章，从实战的角度出发。相关的知识点达到了拿来就用的地步。至此我们就学习完了，可以开心的去睡觉了。