Python使用SQLAlchemy操作MySQL：CRUD完整指南

远山来客

发布时间：2026-10-13 | 3人浏览过

原创

本文详细讲解如何在Python中使用SQLAlchemy库连接MySQL数据库。内容涵盖环境准备、使用ORM创建数据表，以及通过Session对象实现数据的插入、查询、更新和删除等核心CRUD操作，附带完整代码示例与结果分析。

SQLAlchemy简介与准备

SQLAlchemy是一款基于Python的开源软件，提供SQL工具包及对象关系映射（ORM）功能，采用MIT许可证发布。自2006年2月首次发行以来，它迅速成为Python社区中应用最广泛的ORM工具之一，其普及程度不亚于Django自带的ORM框架。

本文将演示如何利用SQLAlchemy对MySQL数据库进行操作，具体包括基础数据表的创建，以及表格数据的增、删、改、查（CRUD）等常用功能实现。

在开始之前，请确保当前Python环境中已安装sqlalchemy和pymysql两个模块。

创建数据表

新建表格

我们以本地MySQL数据库中的orm_test为例，创建一张名为users的表，包含id、name、age和place四个字段。完整的Python代码（create_table.py）如下：


# -*- coding: utf-8 -*-
# author: Jclian91
# place: Sanya, Hainan
# time: 12:46

from sqlalchemy.dialects.mysql import INTEGER, VARCHAR
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()


# users表结构
class Users(Base):
  __tablename__ = 'users'

  id = Column(INTEGER, primary_key=True)
  name = Column(VARCHAR(256), nullable=False)
  age = Column(INTEGER)
  place = Column(VARCHAR(256), nullable=False)


  def __init__(self, id, name, age, place):
    self.id = id
    self.name = name
    self.age = age
    self.place = place

def init_db():
  engine = create_engine(
    "mysql+pymysql://root:@localhost:3306/orm_test",
    encoding= "utf-8",
    echo=True
  )
  Base.metadata.create_all(engine)
  print('Create table successfully!')

if __name__ == '__main__':
  init_db()

执行上述代码后，生成的users表结构如下图所示：

插入数据

数据插入

随后，我们插入几条测试数据。完整的代码（db_insert.py）如下：


# -*- coding: utf-8 -*-
# author: Jclian91
# place: Sanya, Hainan
# time: 12:52

from create_table import Users
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

def insert_data():
  # 初始化数据库连接
  engine = create_engine("mysql+pymysql://root:@localhost:3306/orm_test", encoding="utf-8")
  # 创建DBSession类型
  DBSession = sessionmaker(bind=engine)

  # 创建session对象
  session = DBSession()
  # 插入单条数据
  # 创建新User对象
  new_user = Users(id=1, name='Jack', age=25, place='USA')
  # 添加到session
  session.add(new_user)
  # 提交即保存到数据库
  session.commit()

  # 插入多条数据
  user_list= [Users(id=2, name='Green', age=26, place='UK'),
        Users(id=3, name='Alex', age=31, place='GER'),
        Users(id=4, name='Chen', age=52, place='CHN'),
        Users(id=5, name='Zhang', age=42, place='CHN')
        ]
  session.add_all(user_list)
  session.commit()
  # 关闭session
  session.close()
  print('insert into db successfully!')

if __name__ == '__main__':
  insert_data()

程序运行后，表格中的数据状态如下：

查询数据

数据查询

接下来演示简单的查询操作。我们需要查询place字段为CHN的人员姓名，完整的Python代码（db_query）如下：


# -*- coding: utf-8 -*-
# author: Jclian91
# place: Sanya, Hainan
# time: 12:52

from create_table import Users
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

def query_data():
  # 初始化数据库连接
  engine = create_engine("mysql+pymysql://root:@localhost:3306/orm_test", encoding="utf-8")
  # 创建DBSession类型
  DBSession = sessionmaker(bind=engine)

  # 创建session对象
  session = DBSession()

  # 查询所有place是CHN的人名
  # 创建Query查询，filter是where条件
  # 调用one()返回唯一行，如果调用all()则返回所有行:
  users = session.query(Users).filter(Users.place == 'CHN').all()
  print([use.name for use in users])
  # 输出：['Chen', 'Zhang']

  # 或者用如下查询
  users = session.query(Users.name).filter(Users.place == 'CHN').all()
  print(users)
  # 输出：[('Chen',), ('Zhang',)]

  session.close()


if __name__ == '__main__':
  query_data()

输出结果如下：

['Chen', 'Zhang'] [('Chen',), ('Zhang',)]

更新数据

数据更新

接着演示数据更新功能。我们需要将用户Jack的place字段修改为CHN，完整的Python代码（db_update）如下：


# -*- coding: utf-8 -*-
# author: Jclian91
# place: Sanya, Hainan
# time: 12:52

from create_table import Users
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

def update_data():
  # 初始化数据库连接
  engine = create_engine("mysql+pymysql://root:@localhost:3306/orm_test", encoding="utf-8")
  # 创建DBSession类型
  DBSession = sessionmaker(bind=engine)

  # 创建session对象
  session = DBSession()

  # 数据更新，将Jack的place修改为CHN
  update_obj = session.query(Users).filter(Users.name=='Jack').update({"place":"CHN"})
  session.commit()

  session.close()
  print("Update data successfully!")


if __name__ == '__main__':
  update_data()

更新操作后的表格数据如下：

删除数据

数据删除

最后演示数据删除功能。我们需要删除用户Jack的记录，完整的Python代码（db_delete.py）如下：


# -*- coding: utf-8 -*-
# author: Jclian91
# place: Sanya, Hainan
# time: 12:52

from create_table import Users
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

def delete_data():
  # 初始化数据库连接
  engine = create_engine("mysql+pymysql://root:@localhost:3306/orm_test", encoding="utf-8")
  # 创建DBSession类型
  DBSession = sessionmaker(bind=engine)

  # 创建session对象
  session = DBSession()

  # 数据更新，将Jack的记录删除
  update_obj = session.query(Users).filter(Users.name=='Jack').delete()
  session.commit()

  session.close()
  print("Delete data successfully!")


if __name__ == '__main__':
  delete_data()

程序运行后，表格中的数据状态如下：