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

码农飞哥

发布于 2021-08-18 11:14:10

9860

发布于 2021-08-18 11:14:10

文章被收录于专栏：好好学习

您好，我是码农飞哥，感谢您阅读本文！本文将主要介绍一款应用于Python语言中的ORM框架SQLAlchemy。ORM的是Object-Relational Mapping，作用是把关系数据库的表结构映射到对象上。

安装

本demo使用的是MySQL数据库，ORM框架是SQLAlchemy框架。所以，我们首先需要通过Python的包管理工具安装需要的包。

安装MySQL数据库驱动我们需要安装Python的MySQL驱动来连接MySQL服务器，MySQL官方提供了mysql-connector-python驱动，但是在安装的时候需要给pip命令加上参数 --allow-external。安装命令如下：

pip install mysql-connector-python --allow-external mysql-connector-python

如果该命令安装失败则可以尝试下如下命令：

pip install mysql-connector

安装SQLAlchemy 安装好MySQL驱动之后，接下来就是安装SQLAlchemy模块。同样是通过pip命令安装。

pip install sqlalchemy

环境

软件	版本
Python	3.6.7
mysql-connector-python	8.0.22
SQLAlchemy	1.3.22

使用

MySQL驱动和SQLAlchemy模块安装好之后，我们的准备工作就做完了，就下来就是使用了SQLAlchemy框架了。

1. 待测试的数据表

该脚本创建一个名为job的数据库，然后在该数据库中创建一个名为job_user的表。接着向表中插入两条账号信息。执行该脚本就准备好了测试数据。

CREATE database if NOT EXISTS `job` default character set utf8mb4 collate utf8mb4_unicode_ci;
use `job`;
CREATE TABLE `job_user` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `username` varchar(50) NOT NULL COMMENT '账号',
  `password` varchar(50) NOT NULL COMMENT '密码',
  PRIMARY KEY (`id`),
  UNIQUE KEY `i_username` (`username`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

INSERT INTO `job_user`(`id`, `username`, `password`) VALUES (1, 'admin', '123');
INSERT INTO `job_user`(`id`, `username`, `password`) VALUES (2, 'zhangsan', '345');

2. 导入SQLAlchemy，并初始化DBSession

# 导入相关的包
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 创建对象的基类
Base = declarative_base()
# 初始化数据库连接
engine = create_engine('mysql+mysqlconnector://root:123@localhost:3306/job')
# 创建DBSession类型
DBSession = sessionmaker(bind=engine)
# 定义User对象
class User(Base):
    # 表的名字
    __tablename__ = 'job_user'
    # 表的结构
    id = Column(int(11), primary_key=True)
    username = Column(String(20))
    password = Column(String(20))
#将对象转成dict字典
def to_dict(self):
    return {c.name: getattr(self, c.name, None)
            for c in self.__table__.columns}
Base.to_dict = to_dict

如上代码，主要做了如下三件事情：

导入SQLAlchemy框架的相关类。
初始化数据连接，创建DBSession类型

数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名
mysql+mysqlconnector://root:123@localhost:3306/job

就是说数据库类型是mysql，数据库驱动的名称是mysqlconnector，连接的用户名是root，连接的密码是123，机器地址是：localhost，端口号是：3306，数据库名称是job。这个连接的格式是固定的。接着通过如下语句来创建DBSession类型：

DBSession = sessionmaker(bind=engine)

定义User对象，这个对象里的属性与job_user表里的字段相对应。

3. 插入数据

插入数据的操作其实就是创建一个待插入的User对象，然后将该User对象放入session（会话）中进行提交。

#创建session对象
session = DBSession()
# 创建新User对象
new_user = User(id=3,username='ceshi',password='456')
# 添加到session
session.add(new_user)
# 提交即保存到数据库：
session.commit()
# 关闭session
session.close()

4. 查询数据

根据主键获取记录

根据主键查询可以直接调用get方法，

user = session.query(User).get(1)
print(user and user.to_dict())

就可以直接得到如下结果：

{'id': 1, 'username': 'admin', 'password': '123'}

当然也可以这样写：

user = session.query(User) \
    .filter(User.id == 1) \
    .first()
print(user and user.to_dict())

上面get的方法比较简洁，只适用于主键查询，下面filter的方法比较通用，适用于所有字段的查询。其中first()的意思是返回查询结果中的第一条。

#all()方法是返回所有的查询结果
users = session.query(User) \
    .filter(User.id == 1) \
    .all()
#one()方法是只获取一条记录，如果找不到记录或者多条都会报错
user = session.query(User) \
    .filter(User.id == 1) \
    .one()

AND查询

在实际项目中，我们查询数据通常是多条件查询，那么多条件查询的代码该怎么写呢？

user = session.query(User) \
    .filter(User.username == 'admin',
            User.password > '123') \
    .all()

上面的代码的效果等同于下面的SQL语句;

select * from job_user where username='admin' and password='123'

返回记录条数 count()

#返回记录条数 count()
count = session \
    .query(User) \
    .filter(User.id == 3) \
    .count()
print(count)

将记录按照某个字段进行排序 order_by()

users = session \
    .query(User.id, User.username) \
    .filter(User.id > 2) \
    .order_by(User.id.desc()) \
    .all()

查看记录是否存在 exist()

from sqlalchemy.sql import exists
is_exist = session \
    .query(exists().where(User.id > 10)) \
    .scalar()

OR查询

#需要引入or_函数
from sqlalchemy import or_
users = session.query(User) \
    .filter(or_(User.id == 1,
                User.username =='zhangsan')) \
    .all()

IN查询

users = session.query(User) \
        .filter(User.id.in_([1, 3, 5])) \
        .all()

模糊查询

# 查询用户名包含「zhang」的用户账号
books = session.query(User) \
    .filter(User.username.contains('zhang')) \
    .all()

5. 更新数据

更新数据的操作其实就是首先查出待更新的数据，然后调用update方法更新成新数据。

session = DBSession()
new_user = session.query(User).filter(User.id == 2).update({"username": "lisi"})
session.commit()
session.close()

这里是将id等于2的那条数据的用户名由zhangsan改成lisi。同样的还是先创建session对象，然后提交session,最后关闭session。

6. 删除数据

删除数据与更新数据类似。也是先查出待删除的数据，如果存在则删除，这是物理删除。下面的语句是删除id等于3的用户账号。

# 创建session对象
session = DBSession()
old_user = session.query(User).get(3)
# 删除文件
session.delete(old_user)
# 提交即保存到数据库：
session.commit()
session.close()