深入解析SQLAlchemy:Python数据库CRUD操作最佳实践

一、SQLAlchemy Core 简介

在 Python 的数据库操作领域中，SQLAlchemy Core 是一个极为重要的组成部分。它是 SQLAlchemy 库的核心底层组件，提供了一种灵活且强大的方式来与各种关系型数据库进行交互，包括但不限于 PostgreSQL、MySQL、SQLite 等。

SQLAlchemy Core 的主要功能是允许开发者使用 Python 代码来构建和执行 SQL 语句，同时它还提供了对数据库事务、连接池等关键特性的支持。与直接编写 SQL 语句相比，使用 SQLAlchemy Core 有诸多优势。

可移植性：它实现了数据库的抽象，让开发者可以用统一的代码逻辑操作不同类型的数据库，如 MySQL、PostgreSQL、SQLite 等，极大地提高了代码的可移植性。举个例子，当你需要将项目中的数据库从 SQLite 切换到 MySQL 时，只需要修改连接字符串和部分配置，而核心的数据操作代码无需大幅改动。
安全性：SQLAlchemy Core 通过表达式语言构建 SQL 语句，避免了直接拼接 SQL 字符串，从而有效防止了 SQL 注入攻击，增强了应用程序的安全性。
高性能：它还具备高效的性能，能够在不牺牲太多效率的前提下，提供便捷的数据库操作方式。

二、准备工作

（一）安装 SQLAlchemy

在开始使用 SQLAlchemy Core 进行数据库操作之前，首先需要确保它已经被安装到你的 Python 环境中。SQLAlchemy 是一个第三方库，我们可以使用 pip 这个 Python 包管理工具来进行安装。我们直接使用国内的镜像源，如阿里云、清华大学的镜像源等。以清华大学的镜像源为例，安装命令如下：

pip install sqlalchemy -i https://pypi.tuna.tsinghua.edu.cn/simple

截止目前最新版本是SQLAlchemy==，以下代码都是基于该版本。

（二）安装数据库驱动

1）连接 MySQL 数据库，还需要安装 MySQL 的驱动，如mysql-connector-python或pymysql ，安装命令为：

pip install mysql-connector-python -i https://pypi.tuna.tsinghua.edu.cn/simple
# 或者
pip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple

连接字符串格式为：

# 使用mysql-connector-python驱动
engine = create_engine('mysql+mysqlconnector://username:password@host:port/database')
# 使用pymysql驱动
engine = create_engine('mysql+pymysql://username:password@host:port/database')

2）连接 PostgreSQL 数据库，常用的驱动是psycopg2或pg8000。安装命令如下：

pip install psycopg2 -i https://pypi.tuna.tsinghua.edu.cn/simple
# 或者
pip install pg8000 -i https://pypi.tuna.tsinghua.edu.cn/simple

连接字符串格式为：

# 使用psycopg2驱动
engine = create_engine('postgresql+psycopg2://username:password@host:port/database')
# 使用pg8000驱动
engine = create_engine('postgresql+pg8000://username:password@host:port/database')

3）连接SQLite数据库，它的驱动已经包含在 Python 标准库中，因此不需要额外安装。连接字符串如下：

# 连接到一个名为example.db的SQLite数据库文件
engine = create_engine('sqlite:///example.db')
# 如果使用内存数据库（仅在程序运行期间存在，关闭程序后数据丢失）
engine = create_engine('sqlite:///:memory:')

通过以上步骤，你就完成了 SQLAlchemy 及其相关数据库依赖的安装和环境配置，接下来就可以开始使用 SQLAlchemy 进行数据库操作了。

（三）数据库连接

这里以常用的 MySQL 数据库为例，展示如何使用 SQLAlchemy Core 建立数据库连接。在连接之前，需要准备好数据库的相关信息，包括主机地址、端口号、用户名、密码以及数据库名称。

以下是连接 MySQL 数据库的示例代码：

from sqlalchemy import create_engine
# 数据库连接字符串
DATABASE_URI ='mysql+pymysql://username:password@host:port/database_name'
# 创建数据库引擎
engine = create_engine(DATABASE_URI)
# 使用引擎建立连接
with engine.connect() as conn:
	# 这里可以执行数据库操作
	pass

在上述代码中，create_engine函数用于创建一个数据库引擎，它是与数据库进行交互的核心对象。DATABASE_URI是数据库连接字符串，它包含了连接数据库所需的各种信息：

mysql：表示使用的数据库类型是 MySQL。
pymysql：指定使用的 MySQL 驱动为 pymysql。
username和password：是连接数据库的用户名和密码。
host：是数据库服务器的主机地址。
port：是数据库的端口号。
database_name：是要连接的具体数据库名称。

通过这种方式，我们就可以使用 SQLAlchemy Core 与 MySQL 数据库建立连接，并进行后续的操作。

三、核心功能详解

（一）Table 类

在 SQLAlchemy Core 中，Table类用于表示数据库中的表结构，它是与数据库表进行交互的基础。通过Table类，我们可以定义表的名称、列以及其他相关属性。

参数解释

表名：在创建Table对象时，第一个参数就是表名，是一个字符串，用于指定在数据库中对应的表的名称。例如，Table('users', metadata,...) 中，'users' 就是表名，这个名称要与数据库中实际存在的表名一致，或者在创建表时作为新表的名称。metadata：metadata是元数据对象，它就像是一个容器，用于存储数据库模式的相关信息，包括表、列、约束等。可以把它看作是数据库结构的一个抽象表示。一个metadata对象可以关联多个Table对象。在创建Table时，将metadata传递进去，这样Table对象就与这个元数据对象建立了联系。例如：

from sqlalchemy import MetaData
metadata = MetaData()
users = Table('users', metadata,...)

Column：Column用于定义表中的列，它是Table类的重要参数之一。每个Column对象代表表中的一个列，通过Column可以指定列的名称、数据类型、是否为主键、是否唯一、默认值等属性。例如：

from sqlalchemy import Column, Integer, String
users = Table('users', metadata,
		Column('id', Integer, primary_key=True),
		Column('name', String(), nullable=False),
		Column('age', Integer, default=)
		?)

在上述代码中，定义了一个名为users的表，其中包含三个列：id列，数据类型为Integer，并设置为主键；name列，数据类型为String，长度为，且不允许为空；age列，数据类型为Integer，默认值为。

Schema：指定表所属的数据库模式（schema）。在一些数据库系统（如 PostgreSQL）中，模式用于组织数据库对象，类似于命名空间。

示例：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('postgresql://user:password@localhost/mydb')
metadata = MetaData()
users_table = Table('users',metadata,
                    Column('id', Integer, primary_key=True),
                    Column('name', String),
                    schema='public' # 指定表所属的模式)

quote：如果设置为 True，表名和列名在 SQL 语句中会被引号括起来，这在表名或列名包含特殊字符或关键字时很有用。默认为 False。

示例：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///test.db')
metadata = MetaData()
# 表名包含特殊字符，使用 quote=True
special_table = Table('my-table',metadata,
                      Column('id', Integer, primary_key=True),
                      Column('name', String),
                      quote=True)

autoload_with：用于自动加载表的结构，通过指定数据库引擎来从数据库中获取表的元数据。

示例：

from sqlalchemy import create_engine, MetaData, Table
engine = create_engine('sqlite:///test.db')
metadata = MetaData()
# 自动加载表结构
users_table = Table('users',metadata,
                    autoload_with=engine)
for column in users_table.columns:
	print(column.name)

extend_existing：如果设置为 True，当表已经存在于元数据中时，会扩展现有的表定义，而不是抛出异常。

示例：

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
engine = create_engine('sqlite:///test.db')
metadata = MetaData()
# 第一次定义表
users_table1 = Table('users',metadata,
Column('id', Integer, primary_key=True),
Column('name', String))
# 第二次定义表，使用 extend_existing=True 扩展表定义
users_table2 = Table('users',metadata,
Column('email', String),
extend_existing=True)

2.创建表

使用Table类创建一个简单的数据库表结构，并创建该表。假设我们要创建一个名为books的表，用于存储书籍的信息，包括书籍的 ID、书名、作者和出版年份。

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
# 创建数据库引擎
engine = 'mysql+pymysql://username:password@host:port/testdb'
# 创建元数据对象
metadata = MetaData()
# 使用Table类创建表结构
books = Table('books', metadata,
              Column('id', Integer, primary_key=True),
              Column('title', String), nullable=False),
              Column('author', String(), nullable=False),
              Column('publication_year', Integer)
             )
# 将表结构创建到数据库中
metadata.create_all(engine)

在这个示例中，首先创建了一个 mysql 数据库引擎，连接到名为testdb的数据库。然后创建了metadata元数据对象，接着使用Table类定义了books表的结构，最后通过metadata.create_all(engine)将表结构创建到数据库中。如果数据库中已经存在同名的表，这一步操作会被忽略，不会导致错误。

（二）查询select 类

select类是 SQLAlchemy Core 中用于执行查询操作的核心类，它提供了丰富的功能来构建各种复杂的查询语句。通过select类，我们可以指定查询的列、从哪些表中查询、添加条件过滤、进行分组和排序等操作。

开始以下的查询前，我们先准备一个students表数据：

查询列：在创建select对象时，可以通过参数指定要查询的列。可以是所有列，单个列，或者多个列。例如：

from sqlalchemy import create_engine, MetaData, Table, select
# 创建数据库引擎 数据库信息填入自己的数据表信息
engine = 'mysql+pymysql://username:password@host:port/testdb'

# 创建元数据对象
metadata = MetaData()
# 反射数据库中的 users 表 
users_table = Table('students', metadata, autoload_with=engine)

# 执行查询所有列 users_table.columns或者简写users_table.c 代表所有列
#新版本select对象参数不用传入列表[]
query = select(users_table.columns)

# 执行查询
with engine.connect() as connection:
	result = connection.execute(query)
	for row in result:
		print(row)
# 定义查询某几列 users_table.columns.id表示id列， users_table.columns.age表示age列
query = select(users_table.columns.id,users_table.columns.age)
# 执行查询 
with engine.connect() as connection:
	result = connection.execute(query)
		for row in result:
			print(row)

执行结果：

(2, '李四', , datetime.date, 9, 2))
(3, '王五', , datetime.date, 9, 3))
(4, '刘四', , datetime.date, 9, 2))
(5, '王四', , datetime.date, 9, 2))
(6, '赵六', , datetime.date, 9, 2))
(2, )
(3, )
(4, )
(5, )
(6, )

from 子句：select类会自动根据查询列所在的表确定from子句。如果要显式指定查询的表，可以在创建select对象时传入表对象，主要用于多表join查询。

单表查询示例：

query = select(users_table.columns).select_from(users_table)

通常情况下，如果查询列都来自同一个表，select_from可以省略，SQLAlchemy 会自动推断。但在涉及多表查询时，select_from就非常有用，可以明确指定数据的来源表。

2.多表查询示例：

join代表内连接查询（返回各个表中都有对应关系的数据，是交集数据）
outerjoin代表左外连接（返回左表中的所有记录，以及右表中满足连接条件的记录。如果右表中没有匹配的记录，则右表的字段值为 NULL。）

1）单个join
# 反射数据库中的 users 表和books表
users_table = Table('students', metadata, autoload_with=engine)
books_table = Table('books', metadata, autoload_with=engine)
# 构建查询语句 users_table去join 该books_table
query = select(users_table.columns).select_from(users_table.join(books_table,
                                                                 books_table.c.author == users_table.c.name))
2）多个join
# 反射数据库中的 users 表
users_table = Table('students', metadata, autoload_with=engine)
books_table = Table('books', metadata, autoload_with=engine)
classes_table = Table('classes', metadata, autoload_with=engine)
#多个join
query = select(users_table.columns).select_from(users_table.join(books_table,books_table.c.author == 
                  users_table.c.name).join(classes_table,classes_table.c.name == users_table.c.name))
#多个outerjoin
query = select(users_table.columns).select_from(users_table.outerjoin(books_table,books_table.c.author
             == users_table.c.name).outerjoin(classes_table,classes_table.c.name == users_table.c.name))

where 条件：where条件用于过滤查询结果，只返回符合条件的数据。可以通过where方法添加条件。例如：

# 查询年龄大于的数据
query = select(users_table.columns).where(users_table.columns.age > )
# 查询年龄大于小于的数据
query = select(users_table.columns).where((users_table.columns.age > ) & (users_table.columns.age < ) )
# 查询姓名以'王'开头的用户 使用字符串的startswith方法
query = select(users_table.columns).where(users_table.columns.name.startswith('王'))

使用>、==等比较运算符来构建条件，&用于表示 “且” 的关系，|表示 “或” 的关系。

order_by 排序：order_by用于对查询结果进行排序。可以按照一个或多个列进行升序或降序排序。例如：

from sqlalchemy import desc
# 按照年龄升序排序
query = select(users_table.columns).order_by(users_table.columns.age)
# 按照年龄降序排序 需要使用desc函数
query = select(users_table.columns).order_by(desc(users_table.columns.age))
# 先按照年龄升序，再按照姓名降序排序
query = select(users_table.columns).order_by(users_table.columns.age,desc(users_table.columns.name))

在上述代码中，desc函数用于表示降序排序，如果不使用desc，则默认按照升序排序。

限制查询limit：用于限制查询结果返回的行数。

# 限制返回5行数据
query = select(users_table.columns).limit(5)

分组查询（group by）：

分组查询通常用于统计分析，它可以将数据按照指定的列进行分组，并对每个组应用聚合函数（如sum、count、avg，min，max等）。

from sqlalchemy import select, func
query = select(func.sum(users_table.columns.age).label('age')).group_by(users_table.columns.name)

上面导入的func类，支持sum、count、avg，min，max等聚合函数功能。

多种条件组合查询

将上述提到的where、order_by，group_by、limit等一起使用，就跟我们平常拼接sql一样，如下：

from sqlalchemy import select, func
query = select(users_table.columns.name,func.avg(users_table.columns.age).label('age')).where((users_table.columns.age>)&(users_table.columns.age<)).group_by(users_table.columns.name).order_by(users_table.columns.name).limit(1)

关联查询 in_

有两张表students和books，通过Table类定义表对象后，通过in_做两张表的关联查询，代码如下：查询students表name字段在books表authoor字段有值的数据。类似于我们写sql：select * from students where name in (select author from books)

from sqlalchemy import select
query = select(users_table.columns).where(users_table.columns.name.in_(select(books_table.columns.author)))

（三）插入数据：insert函数

单条插入：

在 SQLAlchemy Core 中，使用insert方法来插入数据。下面是一个向students表中插入单条数据的示例：

from sqlalchemy import create_engine, MetaData, Table, insert
# 创建数据库引擎 数据库信息填入自己的数据表信息
engine = 'mysql+pymysql://username:password@host:port/testdb'
# 创建元数据对象
metadata = MetaData()
# 反射数据库中的表
users_table = Table('students', metadata, autoload_with=engine)
#定义一个字典，存入一条数据，key为表中的列
new_user = {'id':,'name': 'Alice', 'age': }

#通过insert插入上面定义的数据
stmt = insert(users_table).values(new_user)

#执行并提交
with engine.connect() as connection:
	result = connection.execute(stmt)
	connection.commit()

多条插入：

如果要插入多条数据，可以使用insert方法的values参数接受一个数据字典列表。以下是插入多条数据的示例：

#定义一个列表，存入多条条数据字典
new_user = [
{'id':,'name': 'Alice', 'age': },
{'id':,'name': 'Aimy', 'age': }
]
stmt = insert(users_table).values(new_user)
with engine.connect() as connection:
	result = connection.execute(stmt)
	connection.commit()

（四）删除数据：delete函数

使用delete语句可以从数据库表中删除数据。在 SQLAlchemy Core 中，通过delete方法构建删除语句，并使用where条件指定要删除的数据行。

from sqlalchemy import create_engine, MetaData, Table, insert,delete
# 创建数据库引擎 数据库信息填入自己的数据表信息
engine = 'mysql+pymysql://username:password@host:port/testdb'
# 创建元数据对象
metadata = MetaData()
# 反射数据库中的 users 表
users_table = Table('students', metadata, autoload_with=engine)

#删除数据 通过where条件过滤要删除的数据
deldata = delete(users_table).where(users_table.columns.age > )
with engine.connect() as connection:
   result = connection.execute(deldata)
   connection.commit()

（五）更新数据:update函数

在 SQLAlchemy Core 中，通过update方法构建更新语句，并使用where条件指定要更新的数据行，使用values方法指定要更新的字段和值（多个值请传入多个key=value的键值对）

from sqlalchemy import create_engine, MetaData, Table, update
# 创建数据库引擎 数据库信息填入自己的数据表信息
engine = 'mysql+pymysql://username:password@host:port/testdb'
# 创建元数据对象
metadata = MetaData()
# 反射数据库中的 users 表
users_table = Table('students', metadata, autoload_with=engine)

# 更新姓名为王四的用户年龄为
stmt = update(users_table).where(users_table.c.name == '王四').values(age=)

#更新id=9那一行的name和age
stmt = update(users_table).where(users_table.c.id == 9).values(name='李六',age=)

# 建立数据库连接并执行更新操作
with engine.connect() as connection:
   result = connection.execute(stmt)
   connection.commit()

四与直接使用pymysql方式插入万条数据做对比

使用SQLAlchemy和pymysql驱动方式插入万条数据，配合造数神器faker库和时间time库，代码如下：

from sqlalchemy import create_engine, MetaData, Table, insert
from faker import Faker
import time

# 创建数据库引擎
engine = create_engine('mysql+pymysql://xxxxx:xxxxx@xxxxx:/xxxxx)
# 创建元数据对象
metadata = MetaData()
# 反射数据库中的 users 表
users_table = Table('students', metadata, autoload_with=engine)
#定义 Faker对象
fakerins = Faker('zh_CN')

#定义一个数据列表 存入万条数据
new_user = []
id = 
for _ in range):
    new_user.append({'id':id,'name': fakerins.name(), 'age': fakerins.random_int(,)})
    id += 1
#使用insert函数定义插入上述数据
stmt = insert(users_table).values(new_user)
#检查花费时间
start_time = time.time()
with engine.connect() as connection:
   result = connection.execute(stmt)
   connection.commit()
end_time = time.time()
print(f'插入条数据用时{end_time - start_time}s')

执行结果：插入条数据用时6.441025257110596s

直接使用pymysql库批量插入万条数据，同样配合造数神器faker库和时间time库，代码如下：

import pymysql
from faker import Faker
import time
#连接数据库，输入参数为用户名，密码，数据库地址，端口，数据库名 
db =pymysql.connect(user=xxxxx,password=xxxxx,host='xxxxx',port=,database=xxxxx)
#设置数据库游标，数据库操作都是通过游标。
cursor = db.cursor()
#定义Faker对象
fakerins = Faker('zh_CN')
new_user = []
id = 
for _ in range):
   new_user.append((id, fakerins.name(), fakerins.random_int(,)))
   id += 1
#使用executemany方法批量插入数据
start_time = time.time()
cursor.executemany(f"INSERT INTO students(id, name, age) VALUES (%s, %s, %s)",new_user)
db.commit()
end_time = time.time()
cursor.close()
print(f'插入条数据用时{end_time - start_time}s')

执行结果：插入条数据用时2.268887519836426s

总结：在同样的数据库条件下，单次插入万条数据，直接使用pymysql会更高效。

因为SQLAlchemy库需要将 Python 对象转换为数据库能够理解的 SQL 语句，这个过程涉及到对象的映射、属性的检查等操作，多了一些性能开销。

那什么情况下SQLAlchemy库的性能会更优呢？因为我也没涉及到这方面的应用，通过AI搜索了下答案，供大家参考:

共勉：东汉·班固《汉书·枚乘传》：“泰山之管穿石，单极之绠断干。水非石之钻，索非木之锯，渐靡使之然也。”

-----指水滴不断地滴，可以滴穿石头；

-----比喻坚持不懈，集细微的力量也能成就难能的功劳。

----感谢读者的阅读和学习，谢谢大家。

七桃资源网

sqlalchemy 连接数据库 sqlalchemy使用

一、SQLAlchemy Core 简介

二、准备工作

（一）安装 SQLAlchemy

（二）安装数据库驱动

（三）数据库连接

三、核心功能详解

（一）Table 类

参数解释

2.创建表

（二）查询select 类

（三）插入数据：insert函数

（四）删除数据：delete函数

（五）更新数据:update函数

四与直接使用pymysql方式插入万条数据做对比

sqlalchemy 连接数据库 sqlalchemy使用

一、SQLAlchemy Core 简介

二、准备工作

（一）安装 SQLAlchemy

（二）安装数据库驱动

（三）数据库连接

三、核心功能详解

（一）Table 类

参数解释

2.创建表

（二）查询select 类

（三）插入数据：insert函数

（四）删除数据：delete函数

（五）更新数据:update函数

四 与直接使用pymysql方式插入万条数据做对比

四与直接使用pymysql方式插入万条数据做对比