python库sqlalchemy使用教程

前言

SQLAlchemy是一个功能强大的Python ORM工具包。提供全功能的SQL和ORM操作，不再需要编写 SQL 语句，使得操作数据库表就像在操作对象。

安装sqlalchemy

pip install sqlalchemy
pip install pymysql
#如果安装慢的话可以使用清华源
pip install sqlalchemy -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple
1
2
3
4
5

连接数据库

• dialect：数据库，如：sqlite、mysql、oracle等
• driver：数据库驱动，用于连接数据库的，本文使用pymysql
• username：用户名
• password：密码
• host：IP地址
• port：端口
• database：数据库

使用pymysql驱动连接到mysql

from sqlalchemy import create_engine

#mysql+pymysql://:@/[?]
engine = create_engine('mysql+pymysql://user:pwd@localhost/testdb')
1
2
3
4

sqlserver数据库连接示例

# 使用pymssql驱动连接到sqlserver
engine = create_engine('mssql+pymssql://user:pwd@localhost:1433/testdb')
1
2

sqlite数据库连接实例

engine = create_engine('sqlite:///test.db')
1

下面以连接到mysql为例：

HOST = 'localhost'
PORT = 3306
USERNAME = 'root'
PASSWORD = '123456'
DB = 'test'

DB_URI = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB}'
1
2
3
4
5
6
7

建议将配置信息放到你的配置文件中，如config.py

创建ORM模型

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from config import DB_URI

# 初始化数据库连接
engine = create_engine(DB_URI)
Base = declarative_base(engine)  # SQLORM基类
session = sessionmaker(engine)()  # 构建session对象

#定义Student对象
class Student(Base):
	 # 表名
    __tablename__ = 'student'
	
	#表的结构
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50))
    age = Column(Integer)
    sex = Column(String(10))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

新增数据

由于有了ORM，我们向数据库表中添加一行记录，可以视为添加一个Student对象

student = Student(name='Tony', age=18, sex='male')  # 创建一个student对象
session.add(student)  # 添加到session
session.commit()  # 提交到数据库
1
2
3

查询

查询全部数据

item_list = session.query(Student).all()
print(item_list)
for item in item_list:
    print(item.name, item.age)
1
2
3
4

指定查询列

item_list = session.query(Student.name).all()
print(item_list)

# [('Tony',), ('Jane',), ('Ben',)]
1
2
3
4

获取返回数据的第一行

item = session.query(Student.name).first()
print(item)  

# ('Tony',)
1
2
3
4

使用filter()方法进行筛选过滤

item_list = session.query(Student.name).filter(Student.age >= 18).all()
print(item_list)

# [('Tony',), ('Ben',)]
1
2
3
4

使用order_by()进行排序

item_list = session.query(Student.name, Student.age).order_by(Student.age.desc()).all() # desc()表示倒序
print(item_list)

# [('Ben', 20), ('Tony', 18), ('Jane', 16)]
1
2
3
4

多个查询条件（and和or）

# 默认为and, 在filter()中用,分隔多个条件表示and
item_list = session.query(Student.name, Student.age, Student.sex).filter(
    Student.age >= 10, Student.sex == 'female'
).all()
print(item_list)  # [('Jane', 16, 'female')]



from sqlalchemy import or_

# 使用or_连接多个条件
item_list = session.query(Student.name, Student.age, Student.sex).filter(
    or_(Student.age >= 20, Student.sex == 'female')
).all()
print(item_list)  # [('Jane', 16, 'female'), ('Ben', 20, 'male')]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

equal/like/in

# 等于
item_list = session.query(Student.name, Student.age, Student.sex).filter(
    Student.age == 18
).all()
print(item_list)  # [('Tony', 18, 'male')]

# 不等于
item_list = session.query(Student.name, Student.age, Student.sex).filter(
    Student.age != 18
).all()
print(item_list)  # [('Jane', 16, 'female'), ('Ben', 20, 'male')]

# like
item_list = session.query(Student.name, Student.age, Student.sex).filter(
    Student.name.like('%To%')
).all()
print(item_list)  # [('Tony', 18, 'male')]

# in
item_list = session.query(Student.name, Student.age, Student.sex).filter(
    Student.age.in_([16, 20])
).all()
print(item_list) # [('Jane', 16, 'female'), ('Ben', 20, 'male')]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

count计算个数

count = session.query(Student).count()
print(count)  # 3
1
2

切片

item_list = session.query(Student.name).all()[:2]
print(item_list)  # [('Tony',), ('Jane',)]
1
2

修改数据

修改数据可以使用update()方法，update完成后记得执行session.commit()

# 修改Tony的age为22
session.query(Student).filter(Student.name == 'Tony').update({'age': 22})
session.commit()

item = session.query(Student.name, Student.age).filter(Student.name == 'Tony').first()
print(item) 
1
2
3
4
5
6

删除数据

删除数据使用delete()方法，同样也需要执行session.commit()提交事务

# 删除名称为Ben的数据
session.query(Student).filter(Student.name == 'Ben').delete()
session.commit()

item_list = session.query(Student.name, Student.age).all()
print(item_list)
1
2
3
4
5
6

预加载

在 SQLAlchemy 中，预加载是指在查询时一次性加载关联的所有数据，以避免 N+1 查询问题。N+1 查询问题是指在查询数据时，先查询主表，再查询与主表关联的子表，导致查询次数过多，性能下降。预加载技术可以有效解决这个问题。

SQLAlchemy 中提供了一系列预加载方法，常用的有 joinedload() 和 selectinload()。

joinedload() 方法使用 INNER JOIN 将关联表的数据一次性查询出来，然后进行组合，性能较好；selectinload() 方法则会分别查询关联表，然后使用 IN 子句将多个查询结果组合在一起。

预加载还可以用 subqueryload() 方法，它使用子查询的方式一次性加载关联数据。此外，还有 contains_eager() 方法，该方法用于深度预加载，可以预加载多层关联数据。

使用预加载方法，可以显著提高查询效率，减少数据库的 I/O 操作，避免数据丢失和错误。

from sqlalchemy.orm import joinedload, selectinload

# 假设有一个 User 模型和一个 Post 模型，其中 User 模型拥有多篇 Post

# 使用 joinedload 方法预加载 User 和 Post 表
users = session.query(User).options(joinedload(User.posts)).all()

# 使用 selectinload 方法预加载 User 和 Post 表
users = session.query(User).options(selectinload(User.posts)).all()

1
2
3
4
5
6
7
8
9
10

以下是一个简单的示例，演示如何在SQLAlchemy中使用selectinload来优化查询：

from sqlalchemy.orm import selectinload
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String)

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author_id = Column(Integer, ForeignKey('authors.id'))
    author = relationship(Author)

engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# 使用selectinload预加载作者信息
books = session.query(Book).options(selectinload(Book.author)).all()

# 遍历结果集
for book in books:
    print(book.title, book.author.name)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

在上面的示例中，我们定义了两个模型：Author和Book，它们之间通过author_id字段进行关联。我们使用selectinload方法来预加载Book模型中的Author关联，这样我们可以在单个查询中同时获取这两个模型的数据。最后，我们遍历结果集，输出每本书的标题和作者名字。

通过使用selectinload方法，我们可以避免在执行查询时出现N+1查询问题，从而提高应用程序的性能。

一对一查询

例如，假设你有以下两个 ORM 对象，其中 User 对象有一个一对一的关系到 Profile 对象：

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    profile_id = Column(Integer, ForeignKey("profiles.id"))
    profile = relationship("Profile", back_populates="user", uselist=False)

class Profile(Base):
    __tablename__ = "profiles"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="profile", uselist=False)

1
2
3
4
5
6
7
8
9
10
11
12
13

你可以使用以下查询来获取所有用户及其相应的配置文件：

query = session.query(User).options(joinedload("profile"))
1

一对多查询

例如，假设你有以下两个 ORM 对象，其中 User 对象有一个一对多的关系到 Order 对象：

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    orders = relationship("Order", back_populates="user")

class Order(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="orders")
1
2
3
4
5
6
7
8
9
10
11

你可以使用以下查询来获取所有用户及其相应的订单：

query = session.query(User).options(joinedload("orders"))
1

---

• 📢博客主页：https://blog.csdn.net/qq233325332
• 📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正！
• 📢本文由 陌北V1 原创，首发于 CSDN博客🙉
• 📢停下休息的时候不要忘了别人还在奔跑，希望大家抓紧时间学习，全力奔赴更美好的生活✨