本文首發於公眾號：Hunter後端

原文鏈接：SQLAlchemy 連接使用數據庫

相對於 Django 來説，Flask 並沒有 Django 中自帶的那種 ORM 框架，但是我們可以利用第三方的 ORM 框架來進行操作，比如我們這裏介紹的 SQLAlchemy。

接下來這一篇筆記將會介紹如何使用 SQLAlchemy 連接數據庫、建立模型、操作表、以及查詢操作表數據等內容。

以下是本篇筆記目錄：

0. 模塊安裝
1. 數據庫的連接與使用ORM
2. 模型的建立與 ORM Session 的使用
3. 表操作
4. 創建表數據
5. 查詢表數據
6. 更新表數據
7. 刪除表數據

0、模塊安裝

對於 SQLAlchemy 來説，它還需要使用 pymysql 這個庫來進行數據庫的操作，所以這裏我們安裝的模塊如下：

pip3 install sqlalchemy==2.0.19
pip3 install pymysql==1.1.0

1、數據庫的連接與使用

以下是使用 SQLAlchemy 對數據庫進行連接以及一個簡單的查詢示例：

from sqlalchemy import create_engine, text

engine = create_engine("mysql+pymysql://root:123456@192.168.1.5:3306/db_test?charset=utf8")

with engine.connect() as conn:
    sql = "select id, name from users"
    result = conn.execute(text(sql))
    print(result.all())

# [(1, 'admin'), (2, 'user_1')]

在這裏，我們預設在 db_test 這個庫下有一張表名為 users 的表，然後使用原生的 SQL 語句進行了查詢和打印操作，這個操作其實就跟直接使用 pymysql 模塊操作數據庫沒有區別。

如果是插入、更新語句的話，在後面還需要加上 conn.commit() 提交操作。

2、ORM 模型的建立與 ORM Session 的使用

接下來我們定義一個 ORM 模型，其實這裏的 ORM 模型就跟 Django 的模型類似了，不過在語法上有一些區別：

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import Column, Integer, String, Text, DateTime, func

class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), unique=True, nullable=False, comment="用户名")
    email = Column(String(120), default="", comment="郵箱")
    remark = Column(Text, default="", comment="備註信息")
    created_time = Column(DateTime, server_default=func.now(), comment="創建時間")
    updated_time = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment="修改時間")

在這裏，我們定義了一個 Base 類，繼承於 DeclarativeBase，DeclarativeBase 是用於定義 ORM 模型的基類，提供了一些方便的功能，使得使用 ORM 進行數據庫操作更加簡單和直觀，比如提供了一些 query 的方法，這個我們後面再介紹。

對於 User 這個 class，一個類似於 Django 的 ORM 的 SQLAlchemy 的 ORM 模型就搭建完成了。

其中，Integer，String，Text，DateTime 這些都是各自對應到數據庫的字段

對於定義的 Column()，有很多字段的屬性，比如 primary_key 主鍵，auto_increment 自增，default 默認值，nullable 是否允許為 Null 等

在這裏，created_time 和 updated_time 實現的是 Django 的日期時間字段裏 auto_now_add 和 auto_now 屬性，即為創建或者修改的時候值為當前時間

接下來介紹一下 ORM Session 的定義，以下是一個示例：

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

engine = create_engine("mysql+pymysql://root:123456@192.168.1.5:3306/db_test?charset=utf8")

db_session = scoped_session(
    sessionmaker(
        autoflush=False,
        bind=engine
    )
)

上面的操作中，sessionmaker() 通過 engine 創建了一個會話工廠對象，然後通過 scoped_session() 函數創建了一個 scoped_session 對象

我們可以直接使用上面定義出來的 db_session 對數據庫進行操作，也可以對其進行實例化後操作：

db_session.do_something()
// 直接用 db_session 操作數據庫

session = db_session()
session.db_something()
// 用 db_session 實例化結果操作數據庫

他們的使用對象雖然不同，但在功能上都提供了線程本地的會話對象，可以在多線程應用程序中安全地使用。

3、表操作

下面介紹一下 SQLAlchemy 如何操作表。

1. 創建表

在前面我們定義了 engine 和 Base 基類和 User 這個表的 class 之後，我們可以使用下面的方式創建全部定義的表：

Base.metadata.create_all(bind=engine)

執行上面的語句之後，User 這個 class 對應的表就會被創建到數據庫中。

可以把上面這個語句添加到項目的啓動步驟中，因為這個操作會創建庫裏沒有的表，已有的就不會重複執行了。

2. 修改表結構

SQLAlchemy 不支持通過函數的形式直接修改表結構，但是可以執行原生的 SQL 來進行 ALTER TABLE 的操作。

或者通過下面的刪除表操作刪除，再進行 create_all() 操作，但是這樣的話，原表的數據就不存在了。

3. 刪除表

刪除表的操作與創建的方式類似，如下：

Base.metadata.drop_all(bind=engine)

這樣，所有繼承了 Base 基類的表都會被刪除

如果要執行刪除單張表，可以使用 Table 的 drop() 函數：

from sqlalchemy import Table, MetaData

meta_data = MetaData()
table_name = "users"
user_table = Table(table_name, meta_data, autoload_with=engine)
user_table.drop(bind=engine)

4、創建表數據

1. 創建單條數據

u = User(name="admin", email="120@qq.com")
db_session.add(u)
db_session.commit()

使用 add() 添加，然後進行 commit 操作

2. 創建多條數據

u1 = User(name="user_1", email="user1@qq.com")
u2 = User(name="user_2", email="user2@qq.com")
db_session.add_all([u1, u2])
db_session.commit()

批量創建使用 add_all() 函數。

5、查詢表數據

1. 根據主鍵 id 查詢數據

user = db_session.get(User, 1)

返回的 user 就是一個前面我們定義好的 User 對象

2. 條件查詢

條件查詢，可以有兩個操作，一個是 where()，一個是 filter()，這兩者在效果上是相同的，都是作用於條件查詢。

比如，我們要查詢 name 字段的值為 "admin" 以及 id 字段的值為 1 的數據，且返回 id, name, email 字段，可以使用 query() 來進行字段限制，如下操作：

query = db_session.query(User.id, User.name, User.email).filter(User.name == "admin").filter(User.id == 1)

query = db_session.query(User.id, User.name, User.email).filter(User.name == "admin").where(User.id == 1)

query = db_session.query(User.id, User.name, User.email).where(User.name == "admin").filter(User.id == 1)

query = db_session.query(User.id, User.name, User.email).where(User.name == "admin").where(User.id == 1)

如果我們想要一條數據，可以使用 first()，如果想要符合條件的全部數據，可以使用 all()

query.first()
# (1, 'admin', '120@qq.com')

query.all()
# [(1, 'admin', '120@qq.com')]

除此之外，我們還可以將 db_session.query_property() 賦值給 Base.query，後面就可以直接通過 User 進行查詢操作

Base.query = db_session.query_property()

User.query.filter(User.name == "admin").filter(User.id == 1)

然後通過 first() 或者 all() 返回的就是定義的 User 的對象

6、更新表數據

對於我們獲取到的 ORM 模型實例，比如我們在前面通過主鍵 id 獲取到的數據，或者在 db_session.query() 中不指定字段，直接指定模型獲取到的數據，我們可以直接對其字段進行修改，然後 commit

user_1 = db_session.get(User, 1)
user_1.email = "999@qq.com"
db_session.add(user_1)

user_2 = db_session.query(User).filter(User.id == 2).first()
user_2.email = "888@qq.com"
db_session.add(user_2)

db_session.commit()

或者我們前面的 query 屬性的方式也可以：

User.query.filter(User.id == 1).update({"email": "19283@qq.com"})
db_session.commit()

7、刪除表數據

1. 單條記錄刪除

user = db_session.get(User, 2)
db_session.delete(user)
db_session.commit()

2. 根據條件批量刪除

User.query.filter(User.id == 3).delete()
db_session.commit()

如果想獲取更多後端相關文章，可掃碼關注閲讀：