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

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

SQLAlchemy的安装

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

#指定阿里镜像服务器安装SQLAlchemypip 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 randomimport time, datetimefrom typing import Listfrom typing import Optionalfrom sqlalchemy import create_enginefrom sqlalchemy import ForeignKeyfrom sqlalchemy import String, DateTime, Integerfrom sqlalchemy import func, textfrom sqlalchemy.orm import DeclarativeBasefrom sqlalchemy.orm import Mappedfrom sqlalchemy.orm import mapped_columnfrom sqlalchemy.orm import relationshipfrom sqlalchemy.orm import Sessionfrom sqlalchemy import selectfrom sqlalchemy.orm import sessionmakerclass Base(DeclarativeBase):    passclass 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