Python ORM 框架 SQLAlchemy 的简单使用_python orm工具

目录

 

一·SQLAlchemy介绍

二·概念与数据类型预习

 三·使用步骤

1.安装依赖包

2.创建连接

3.创建实体对象

4.创建数据表

 5.数据操作

四·工具方法封装

增删改查代码封装

实体类修改构造方法

工具的使用：

---

一·SQLAlchemy介绍

SQLAlchemy 是 Python SQL工具集和对象关系映射框架(ORM)，为开发者提供了强大且灵活的数据库操作。

他提供了一整套著名的企业级持久化存储模式，具有高效、高性能的数据库访问设计，并适配于简单易用的 Python 语言。

上述描述来自SQLAlchemy官方网站，更多详情请查看链接

二·概念与数据类型预习

 

 三·使用步骤

1.安装依赖包

# 安装mysql数据库Python依赖包
pip install pymysql
 
# 安装ORM框架
pip install SQLAlchemy

2.创建连接

这里假设使用的是本地的MySQL数据库，数据库名 test， 数据表名 STUDENTS。

from sqlalchemy import create_engine
 
# 数据使用本地，用户名root, 密码123456，数据库名test,密码123456(我随便取的，开发时一定注意密码的复杂度问题)
db_link = f'mysql+pymysql://root:123456@127.0.0.1:3306/test'
 
# 获取数据库连接对象
engine = create_engine(db_link)
 
 
engine = create_engine(db_link,
                       echo=True,
                       pool_size=8,
                       pool_recycle=60*30
                       )
# echo: 当设置为True时会将orm语句转化为sql语句打印，一般debug的时候可用
# pool_size: 连接池的大小，默认为5个，设置为0时表示连接无限制
# pool_recycle: 设置时间以限制数据库多久没连接自动断开

3.创建实体对象

ORM的重要特点，就是我们操作表的时候就需要通过操作对象来实现，现在我们来创建一个类，以常见的学生表举例：

from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
 
# 创建对象的基类:
Base = declarative_base()
 
 
# 定义User对象:
class Student(Base):
    # 表的名字:
    __tablename__ = 'STUDENTS'
 
 
 
    def __init__(self, name, age, gender):
        self.name = name
        self.age = age
        self.gender = gender
 
    # 表的结构:
    id = Column(name='stu_id', type_=Integer, primary_key=True, autoincrement=True)
    name = Column(name='stu_name', type_=String(20))
    age = Column(name='stu_age', type_=Integer)
    gender = Column(name='stu_gender', type_=String(20))

declarative_base()是sqlalchemy内部封装的一个方法，通过其构造一个基类，这个基类和它的子类，可以将Python类和数据库表关联映射起来。

数据库表模型类通过__tablename__和表关联起来，Column表示数据表的列。

4.创建数据表

如数据表已通过执行SQL语句创建，则忽略此步骤

# 创建数据表
Base.metadata.create_all(engine)

执行结果如图所示

 

 5.数据操作

获取session会话

sqlalchemy中使用session对象用于创建程序和数据库之间的会话，所有对象的载入和保存都需要通过session对象 。

通过sessionmaker调用创建一个session工厂，并关联Engine，以确保每个session都可以使用该Engine连接资源：

from sqlalchemy.orm import sessionmaker
 
# 创建session工厂
session_factory= sessionmaker(bind=engine)
 
# 创建 session 对象
session = session_factory()

session对象的常见操作方法包括：

1. flush：预提交，提交到数据库文件，还未写入数据库文件中
2. commit：事务提交
3. rollback：事务回滚
4. close：关闭会话

新增数据

# 创建一个学生对象
new_student = Student('John', 18, "Male")
# 将新数据插入到数据库表中
session.add(new_student)
# 事务提交
session.commit()

查询数据

简单数据查询，例如查询所有的学生数据：

session.query(Student)

需要注意的是，这里只构造Query，事实上并没有发送至数据库进行查询，只会在Query.get()、Query.all()、Query.one()以及Query.__iter__等具有“执行”语义的函数，才会真的去获取.所以完整的语句是：

all_students = session.query(Student).all()
print(len(all_students))
 
# 由于数据库中只有一条数据，所以此处输出数据是 1

单数据表条件查询，例如，查询编号为1的学生信息

# 单数据表条件查询，查询id为1的学生
criteria_students = session.query(Student).filter_by(id=1).all()
print(len(criteria_students))
 
# 或
 
# 单数据表条件查询，查询id为1的学生
criteria_students = session.query(Student).filter(Student.id == 1).all()
print(len(criteria_students))

query有filter和filter_by两个过滤方法，通常这两个方法都会用到的，所以一定要掌握它们的区别：

修改数据

数据修改操作同样有两种方式，批量修改或同时需要修改多个属性时，例如同时修改age和gender(举个栗子)，直接使用第一种方法传入字典数据即可{ "age" : 20, "gender" : "Female" }

# 第一种修改方式
session.query(Student).filter_by(id=1).update({"age": 20})
session.commit()
 
# 第二种修改方式
stu = session.query(Student).filter_by(id=1).first()
stu.age = 20
session.add(stu)
session.commit()

删除数据

# 第一种删除方式
del_stu = session.query(Student).filter_by(id=1).first()
if del_stu:
    session.delete(del_stu)
    session.commit()
 
 
# 第二种删除方式
session.query(Student).filter_by(id=1).delete()
session.commit()

以上就是SQLAlchemy的简单增删改查操作了。示例源码放在下面：

# -*- coding: UTF-8 -*-
from sqlalchemy import create_engine
 
from student import Base, Student
from sqlalchemy.orm import sessionmaker
 
db_link = f'mysql+pymysql://root:aoto123@192.168.1.124:3306/test'
engine = create_engine(db_link)
 
print(engine)
 
# 创建数据表
# Base.metadata.create_all(engine)
 
# 创建session
DbSession = sessionmaker(bind=engine)
session = DbSession()
 
# 创建一个学生对象
# new_student = Student('John', 18, "Male")
# 将新数据插入到数据库表中
# session.add(new_student)
# 事务提交
# session.commit()
 
# 查询所有的学生数据
all_students = session.query(Student).all()
print(len(all_students))
 
# 单数据表条件查询，查询id为1的学生
# criteria_students = session.query(Student).filter_by(id=1).all()
# print(len(criteria_students))
 
# 单数据表条件查询，查询id为1的学生
# criteria_students = session.query(Student).filter(Student.id == 1).all()
# print(len(criteria_students))
 
# 第一种修改方式
# session.query(Student).filter_by(id=1).update({"age": 20})
# session.commit()
 
# 第二种修改方式
# stu = session.query(Student).filter_by(id=1).first()
# stu.age = 20
# session.add(stu)
# session.commit()
 
# 第一种删除方式
# del_stu = session.query(Student).filter_by(id=1).first()
# if del_stu:
#     session.delete(del_stu)
#     session.commit()
 
# 第二种删除方式
# session.query(Student).filter_by(id=1).delete()
# session.commit()

四·工具方法封装

增删改查代码封装

上述代码中关于Student的数据增删改查，类与查询条件的耦合度较高，用于项目中大量使用的话会造成代码比较混乱，这里分享一个简单封装的操作类，水平一般，仅供参考

# -*- coding: UTF-8 -*-
 
class DatabaseUtil(object):
 
    def __init__(self, session):
        """
        初始化工具类
        :param session: 数据库连接会话
        """
        self.db_session = session
 
    def add_obj(self, obj):
        """
        插入数据
        :param obj: 插入到数据库表记录的映射对象
        :return:
        """
        self.db_session.add(obj)
        # 提交即保存到数据库:
        self.db_session.commit()
 
    def del_obj(self, class_, *args):
        """
        删除数据库记录
        :param class_: 数据表所对应的对象类型
        :param args: 筛选条件，元组类型
        :return:
        """
        self.db_session.query(class_).filter(args[0] == args[1]).delete()
        self.db_session.commit()
 
    def is_obj_exists(self, class_, *args):
        """
        查询数据库中是否有数据记录
        :param class_: 数据表所对应的对象类型
        :param args: 筛选条件，元组类型
        :return: 有记录，返回True，否则返回False
        """
        obj_list = self.db_session.query(class_).filter(args[0] == args[1]).all()
        if len(obj_list) > 0:
            return True
        else:
            return False
 
    def insert_or_update(self, class_, *args, **kwargs):
        """
        插入或更新一条记录
        :param class_: 数据表所对应的对象类型
        :param args: 筛选条件，元组类型
        :param kwargs: 插入到数据库中的数据记录，字典类型
        :return:
        """
        if self.is_obj_exists(class_,  *args):
            # update
            self.db_session.query(class_).filter(args[0] == args[1]).update(kwargs)
            self.db_session.commit()
        else:
            # insert
            obj = class_(**kwargs)
            self.add_obj(obj)
 
 
    @staticmethod
    def props(obj):
        """
        将class转dict,以_开头的属性不要
        :param obj:
        :return:
        """
        pr = {}
        for name in dir(obj):
            value = getattr(obj, name)
            if value is None:
                continue
            if not name.startswith('_') and not callable(value):
                pr[name] = value
        return pr

那么这个工具类如何使用呢，这里有个使用示例，请注意，这个工具类在初始化时，直接接收session值作为属性成员，您也可以在工具类直接接收数据库连接参数等数据，自己生成session会话。

实体类修改构造方法

在使用前，修改一下实体类的构造方法，直接转字典类型为class类型，方便操作，如下：

from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
 
# 创建对象的基类:
Base = declarative_base()
 
 
# 定义User对象:
class Student(Base):
    # 表的名字:
    __tablename__ = 'STUDENTS'
 
    def __init__(self, **entries):
        self.__dict__.update(entries)
 
    # 表的结构:
    id = Column(name='stu_id', type_=Integer, primary_key=True, autoincrement=True)
    name = Column(name='stu_name', type_=String(20))
    age = Column(name='stu_age', type_=Integer)
    gender = Column(name='stu_gender', type_=String(20))

工具的使用：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
 
from database_test.db_utils import DatabaseUtil
from student import Student
 
if __name__ == '__main__':
    stu_info_1 = {"name": "John", "age": 18, "gender": "Male"}
    stu_info_2 = {"name": "Gina", "age": 20, "gender": "Female"}
    stu_info_3 = {"name": "Tom", "age": 23, "gender": "Male"}
 
    db_link = f'mysql+pymysql://root:aoto123@192.168.1.124:3306/test'
    engine = create_engine(db_link)
 
    # 创建session
    DbSession = sessionmaker(bind=engine)
    session = DbSession()
 
    # 获取工具对象
    db_util = DatabaseUtil(session)
    # 字典数据转换为Student对象
    new_stu = Student(**stu_info_1)
    # 新增
    db_util.add_obj(new_stu)
 
    # 新增或更新数据
 
    # 设置查询条件，等同于 where name = "some name" 和 query(Student).filter(name='some name')
    criteria = (Student.name, stu_info_1["name"])
    # 传入参数，如果这条记录不存在，插入数据，如果已存在，按照传入的数据更新
    db_util.insert_or_update(Student, *criteria, **stu_info_1)
 
    # 删除数据
    # 设置查询条件，等同于delete ...  where name = "some name"
    db_util.del_obj(Student, *criteria)
 
    exit(0)

以上代码仅供参考，水平有限，欢迎各位批评指正。