最近一段时间全部花在研究微信逆向技术上了，就一直没有写总结，历时一个月终于搞定了微信搜一搜逆向的问题，就赶紧补上一篇总结，关于微信逆向技术也攒了一些经验，后面慢慢地整理成文。

SQLAlchemy是python语言中一种非常知名的ORM库，使用ORM库的目的是使开发人员不必直接操作繁琐的sql，而是以操作对象的方式来操作数据库。我自己有过这样的体会，一开始我以为只是简单的几张表而已，没有什么复杂的操作，直接写sql就好了，而且之前也没有使用过ORM库，不愿意花费额外的精力去学习它。可是随着功能的变动，特别是一段时间之后再次拿起原来的代码时，我需要花费较多的时间去理清sql干了什么，这增加了我重新理解项目的复杂度，让我的大脑感觉非常费力，改动起来也特别容易出错。之后强迫自己花了点额外的时间掌握ORM库之后，发现很少再有类似的困惑了，它带来的收益是显著的。本文介绍SQLAlchemy的基本使用，之后针对SQLite数据库的更新性能进行了简单的测试，对比了三种不同的策略带来的性能提升情况。

SQLAlchemy的安装

执行pip install sqlalchemy命令进行安装，因为python官方库服务器在国外，有时候会比较慢，可以指定阿里服务器镜像进行安装，我本机安装后的版本为2.0.7。

#指定阿里镜像服务器安装SQLAlchemy
pip install sqlalchemy -i https://mirrors.aliyun.com/pypi/simple/

>>> import sqlalchemy
>>> sqlalchemy.__version__
'2.0.7'

数据库建模

本文使用如下两张表进行演示

• dept为部门信息表，包含id(主键)、name、update_at字段
• employee为员工信息表，包含id(主键)、name、age、sex、dept_id、update_at字段，其中dept_id为外键，对应dept表的id，dept和employee为一对多的关系，表示部门下可以有多名员工

SQLAlchemy为两张表建模的代码如下：

import random
import time, datetime
from typing import List
from typing import Optional
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import String, DateTime, Integer
from sqlalchemy import func, text
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy import select
from sqlalchemy.orm import sessionmaker

class Base(DeclarativeBase):
    pass

class Department(Base):
    # 表名dept为部门信息表
    __tablename__ = "dept"
    # id为主键
    id: Mapped[int] = mapped_column(primary_key=True)
    # name为部门名称字段
    name: Mapped[str] = mapped_column(String(100))
    # update_at字段表示更新时间，默认值为当前系统时间
    update_at: Mapped[DateTime] = mapped_column(DateTime(), default=datetime.datetime.now)
    # employees对应该指定部门下的员工信息列表，back_populates表示反向关联的表名为自身，
    # cascade指定为级联删除，即删除部门信息会一并删除部门下的员工信息
    employees: Mapped[List["Employee"]] = relationship(back_populates="dept", cascade="all, delete-orphan")

    # 字符串形式表示Department对象
    def __repr__(self) -> str:
        return f"Department(id={self.id!r}, name={self.name!r})"

class Employee(Base):
    # 表名employee为员工信息表
    __tablename__ = "employee"
    # id为主键
    id: Mapped[int] = mapped_column(primary_key=True)
    # name为员工姓名
    name: Mapped[str] = mapped_column(String(100))
    # age为员工年龄
    age: Mapped[int] = mapped_column(Integer)
    # sex为员工性别
    sex: Mapped[str] = mapped_column(String(10))
    # dept_id为外键，对应dept表的id字段
    dept_id: Mapped[int] = mapped_column(ForeignKey("dept.id"))
    # update_at字段表示更新时间，默认值为当前系统时间
    update_at: Mapped[DateTime] = mapped_column(DateTime(), default=datetime.datetime.now)
    # dept和Department中的employees必须成对出现，否则会报错，表示两个表之间的关系back_populates对应employees，而不是表名employee
    dept: Mapped["Department"] = relationship(back_populates="employees")

    # 字符串形式表示Department对象
    def __repr__(self) -> str:
        return f"Employee(id={self.id!r}, name={self.name!r}, dept_id={self.dept_id!r})"

连接数据并建表

SQLAlchemy中engine对象负责使用DBAPI来适配不同类型的数据库。SQLite是应用非常广泛的轻量型数据库引擎，它的数据库就是一个文件，完全无需配置，它可以满足绝大部分中小型网站的使用，微信中也使用了SQLite数据库，这里也以SQLIte数据库为例进行介绍。
以下代码将在python脚本同级目录下创建demo.db数据库文件，并根据上面定义的模型创建出表dept和employee。

# 创建engine对象
engine = create_engine("sqlite:///demo.db", echo=False)
# 建立表
Base.metadata.create_all(engine)

增删改查

定义了模型并创建好表之后，就可以进行增删改查操作了，不过需要事先创建Session对象，它类似数据库中的事务操作对象

# 创建session，类似于数据库中的事务
Session = sessionmaker(bind=engine)
session = Session()

下面给出基本的增删改查示例代码，代码很容易看懂，可以每种方法单独进行测试并通过可视化工具查看执行结果，这里不再赘述

def insert_demo():
    """插入表数据示例
    """
    # 部门信息记录
    dept_1 = Department(name="研发部")
    if session.query(Department).filter_by(name="研发部").first() is None:
        session.add(dept_1)
    dept_2 = Department(name="营销部")
    if session.query(Department).filter_by(name="营销部").first() is None:
        session.add(dept_2)
    session.commit()
    # 员工信息记录
    employee_list = []
    empoyee_1 = Employee(name="张三", age=20, sex="男", dept_id=dept_1.id)
    if session.query(Employee).filter_by(name="张三").first() is None:
        employee_list.append(empoyee_1)
    empoyee_2 = Employee(name="张小丽", age=23, sex="女", dept_id=dept_2.id)
    if session.query(Employee).filter_by(name="张小丽").first() is None:
        employee_list.append(empoyee_2)
    empoyee_3 = Employee(name="黎明", age=30, sex="男", dept_id=dept_2.id)
    if session.query(Employee).filter_by(name="黎明").first() is None:
        employee_list.append(empoyee_3)
    session.add_all(employee_list)
    session.commit()

def query_demo():
    """查询数据示例
    """
    dept_1 = session.query(Department).filter_by(name="研发部").first()
    print(f"查询部门名称为:{dept_1.name}")
    print(f"该部门下的员工有：{dept_1.employees}, 数量为{len(dept_1.employees)}")

def update_demo():
    """更新数据示例
    """
    employees = session.query(Employee).all()
    for emp in employees:
        # 年龄增加1
        emp.age += 1
        emp.update_at = datetime.datetime.now()
    session.commit()

def delete_demo():
    """删除数据示例
    """
    dept_1 = session.query(Department).filter_by(name="研发部").first()
    # 该部门名下的员工记录也会被删除
    session.delete(dept_1)
    session.commit()

SQLite性能验证

最后，我们来测试一下SQLite数据库更新1万条员工数据的性能情况，分别针对逐条查询员工信息逐条提交、逐条查询员工一次性提交、一次性查询员工一次性提交三种情况进行测试。

def perf_test():
    """插入性能测试
    """
    # 插入1万条员工记录,先删除记录确保可以重复执行
    employees = session.query(Employee).filter(Employee.id >= 10).all()
    for emp in employees:
        session.delete(emp)
    session.commit()
    dept_1 = session.query(Department).filter_by(name="营销部").first()
    for i in range(10, 10010):
        emp = Employee(id=i, name=f"emp_{i}", age=random.randint(18, 45), sex=random.choice(["男","女"]), dept_id=dept_1.id)
        session.add(emp)
    session.commit()

    # 1.逐条查询员工记录、逐条提交性能测试结果
    start = time.perf_counter()
    for i in range(10, 10010):
        emp = session.query(Employee).filter_by(id=i).first()
        emp.name = f"emp_update_{i}"
        session.commit()
    print(f"逐条查询、逐条提交方式更新10000条记录共耗时{time.perf_counter()-start}秒。")
    # 2.逐条查询员工记录，统一提交一次性能测试结果
    start = time.perf_counter()
    for i in range(10, 10010):
        emp = session.query(Employee).filter_by(id=i).first()
        emp.name = f"emp_update_{i}"
    session.commit()
    print(f"逐条查询、一次性提交方式更新10000条记录共耗时{time.perf_counter()-start}秒。")

    # 3.员工信息一次性读入内存，修改员工信息，统一提交一次性能测试结果
    start = time.perf_counter()
    employees = session.query(Employee).filter(Employee.id >= 10).all()
    for emp in employees:
        emp.name = f"emp_update_{i}"
    session.commit()
    print(f"一次性查询到内存、一次性提交方式更新10000条记录共耗时{time.perf_counter()-start}秒。")

if __name__ == "__main__":
    # 创建engine对象
    engine = create_engine("sqlite:///demo.db", echo=False)
    # 建立表
    Base.metadata.create_all(engine)
    # 创建session，类似于数据库中的事务
    Session = sessionmaker(bind=engine)
    session = Session()
    # 增删改查测试
    insert_demo()
    query_demo()
    update_demo()
    delete_demo()
    # 性能测试
    perf_test()

最终测试结果如下，从结果可以看出一次性将员工信息读到内存，更新之后再一次性提交的方式是最快的，性能是最差情况的200倍以上，在实际使用中可以根据情况采取合适的方式进行优化。

查询部门名称为:研发部
该部门下的员工有：[Employee(id=10010, name='张三', dept_id=3)], 数量为1
逐条查询、逐条提交方式更新10000条记录共耗时115.6898583秒。
逐条查询、一次性提交方式更新10000条记录共耗时12.288941499999993秒。
一次性查询到内存、一次性提交方式更新10000条记录共耗时0.520228799999984秒。

参考文献

[1] Overview — SQLAlchemy 2.0 Documentation