3個(gè)Python?SQLAlchemy數(shù)據(jù)庫操作功能詳解

更新時(shí)間：2023年09月12日 10:37:07 作者：天行無忌

Python?SQLAlchemy?是一個(gè)強(qiáng)大且多功能的?Python?SQL?工具包和對(duì)象關(guān)系映射?(ORM)?系統(tǒng),提供了一整套眾所周知的企業(yè)級(jí)持久性模式,本文為大家整理了它必須了解的3個(gè)數(shù)據(jù)庫操作功能,希望對(duì)大家有所幫助

Python SQLAlchemy 是一個(gè)強(qiáng)大且多功能的 Python SQL 工具包和對(duì)象關(guān)系映射 (ORM) 系統(tǒng)，提供了一整套眾所周知的企業(yè)級(jí)持久性模式，專為高效和高性能的數(shù)據(jù)庫訪問而設(shè)計(jì)。它可以處理從小型簡(jiǎn)單查詢到高負(fù)載場(chǎng)景中復(fù)雜事務(wù)的所有事務(wù)。它為開發(fā)人員提供了一個(gè)高級(jí)的 Pythonic 接口來與關(guān)系數(shù)據(jù)庫交互，允許使用 Python 類和對(duì)象來處理數(shù)據(jù)庫，而不是編寫復(fù)雜的 SQL 查詢，并以這種方式抽象代碼，刪除所有支持的數(shù)據(jù)庫腳本語句，例如 PostgreSQL、MySQL、SQLite 和 Oracle 等等。許多流行的 Python 框架都使用 SQLAlchemy，例如 Django、Flask 和 Pyramid。

本文一起來學(xué)習(xí) SQLAlchemy 庫的三個(gè)有用功能，在大部分的項(xiàng)目開發(fā)中都可能用得上的功能。

下面先來了解一些基礎(chǔ)知識(shí)。

基礎(chǔ)知識(shí)

從這里開始介紹一些基本的知識(shí)，對(duì)于大多數(shù)人來說基本都是熟悉的知識(shí)。

數(shù)據(jù)模型

本文將以一個(gè)保存用戶記錄的簡(jiǎn)單 SQL 表作為實(shí)例數(shù)據(jù)，代碼將重點(diǎn)關(guān)注 SQLAlchemy 集成，以實(shí)現(xiàn)對(duì)數(shù)據(jù)存儲(chǔ)的需求，不涉及 Web 框架或任何其他復(fù)雜性。

為示例定義的基本用戶模型如下：

from pydantic import BaseModel
from typing import Optional
class UserSignUp(BaseModel):
    name: str
    surname: Optional[str] = None
    birth_year: Optional[int] = None
    notes: Optional[str] = None

在上面的代碼片段中，定義了一個(gè) Pydantic 模型，它是想要在數(shù)據(jù)庫中保存記錄的數(shù)據(jù)的模型。Pydantic 是 Python 中最新且最有前途的庫，用于處理數(shù)據(jù)驗(yàn)證、結(jié)構(gòu)和建模。如果不熟悉它，其實(shí)跟 Mongoose 、Prisma 實(shí)現(xiàn)原理相似。

存儲(chǔ) Storage

出于存儲(chǔ)目的，將使用最簡(jiǎn)單的 SQL 數(shù)據(jù)庫系統(tǒng)之一，即 SQLite。為此，需要定義一個(gè)引擎來與文件系統(tǒng)中實(shí)際的 SQLite 數(shù)據(jù)庫文件進(jìn)行通信，并創(chuàng)建一個(gè)會(huì)話對(duì)象，以便可以與數(shù)據(jù)庫進(jìn)行交互以進(jìn)行操作。

from datetime import datetime
from pydantic import BaseModel
import os
import json
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy.sql import func
from sqlalchemy import Column, DateTime, Integer, String
project_dir = os.path.dirname(os.path.abspath(__file__))
engine = create_engine("sqlite:///{}".format(os.path.join(project_dir, "./storage.db")),
                       connect_args={"check_same_thread": False})
Base = declarative_base()
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())
    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        return json.dumps(dict_repr, indent=2)
Base.metadata.create_all(bind=engine)

在上面的代碼片段中，項(xiàng)目文件夾路徑用于指向數(shù)據(jù)庫路徑。數(shù)據(jù)庫名稱定義為 storage.db，如果不存在，則會(huì)自動(dòng)創(chuàng)建。

此外，使用 SQLAlchemy ORM 模塊的聲明函數(shù)來構(gòu)造一個(gè)基類，然后由 User 類來使用，User 類表示數(shù)據(jù)庫中保存用戶記錄的表。

表用戶行被定義為包含比我們?yōu)橛脩糇?cè)引入的數(shù)據(jù)模型額外的兩個(gè)字段。

User 表包含注冊(cè)引入的數(shù)據(jù)模型額外的兩個(gè)字段。

一個(gè)自動(dòng)增量整數(shù)，作為已索引并用作表主鍵的行的 ID。
注冊(cè)日期字段，它將當(dāng)前時(shí)間戳作為默認(rèn)參數(shù)，這意味著正在數(shù)據(jù)庫中寫入的條目的時(shí)間戳。

數(shù)據(jù)庫表定義類還包含一個(gè)額外的魔術(shù)方法，__repr__ 方法在此處定義以實(shí)現(xiàn)自定義序列化器。這樣做的目的是可以直接打印數(shù)據(jù)庫記錄，即數(shù)據(jù)記錄行，因?yàn)樗鼈兪鞘褂?SQLAlchemy 庫從表中獲取的。

最后，通過調(diào)用模塊 create_all 的方法，Base.metadata 在創(chuàng)建的數(shù)據(jù)庫中實(shí)例化定義的表結(jié)構(gòu)。因此，列及其數(shù)據(jù)類型和主鍵定義現(xiàn)在都存在。

寫入數(shù)據(jù)庫表

定義并創(chuàng)建了數(shù)據(jù)庫表，現(xiàn)在來嘗試寫入一條記錄。首先使用之前創(chuàng)建的引擎對(duì)象來創(chuàng)建一個(gè)到數(shù)據(jù)庫的會(huì)話。

from sqlalchemy import create_engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker
session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db_session = session_local()

然后，使用聲明的數(shù)據(jù)模型創(chuàng)建一個(gè)要注冊(cè)的新用戶。

user_sign_up = UserSignUp(
    name="Quintion",
    surname="Tang",
    notes="一個(gè)工程師"
)

最后，使用 Users 的數(shù)據(jù)庫表類創(chuàng)建一個(gè)條目，以使用建立的會(huì)話寫入數(shù)據(jù)庫并提交更改。

try:
    user_to_store = User(
        name=user_sign_up.name,
        surname=user_sign_up.surname,
        birth_year=user_sign_up.birth_year,
        notes=user_sign_up.notes
    )
    db_session.add(user_to_store)
    db_session.commit()
    print(user_to_store)
except IntegrityError as e:
    db_session.rollback()
    print(f"用戶注冊(cè)時(shí)出錯(cuò)：{str(e)}")

如果沒有因任何完整性錯(cuò)誤引發(fā)異常，則應(yīng)寫入該記錄并將其提交到數(shù)據(jù)庫。

{
  "id": 1,
  "name": "Quintion",
  "surname": "Tang",
  "birth_year": null,
  "notes": "一個(gè)工程師",
  "register_date": "2023-09-10T14:29:50"
}

請(qǐng)注意，birth_year 字段為空，因?yàn)槭紫葲]有為其提供任何整數(shù)，并且在寫入操作期間會(huì)自動(dòng)為 register_date 生成時(shí)間戳。

數(shù)據(jù)驗(yàn)證和預(yù)處理

在實(shí)際用例中，在存儲(chǔ)之前，首先都需要預(yù)先驗(yàn)證或處理數(shù)據(jù)。

來看看將如何在這里做到這一點(diǎn)。例如，假設(shè)想要將每個(gè)字符串字段大寫。如果需要在數(shù)據(jù)庫中對(duì)數(shù)據(jù)進(jìn)行標(biāo)準(zhǔn)化，這樣的過程將對(duì)于后期數(shù)據(jù)處理分析非常有用。統(tǒng)一字符串字段大寫的好處是在后續(xù)相關(guān)功能（如檢索）不必?fù)?dān)心區(qū)分大小寫的問題。

那么在 SQLAlchemy 中是如何使用其 ORM validates 裝飾器來完成此操作。

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())
    # 這里對(duì)數(shù)據(jù)進(jìn)行預(yù)處理
    @validates('name', 'surname', 'notes')
    def normalize(self, _, value):
        if type(value) == str:
            return value.upper()
    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        return json.dumps(dict_repr, indent=2)

請(qǐng)注意，驗(yàn)證裝飾器接受定義為字符串?dāng)?shù)據(jù)類型列的所有三列的參數(shù) @validates('name', 'surname', 'notes')。當(dāng)然，這不是強(qiáng)制性的，可以使用任意數(shù)量的列并選擇不同的操作進(jìn)行預(yù)處理。

函數(shù) normalize 將應(yīng)用于所有選定的三列，傳入數(shù)據(jù)將通過此功能將要寫入數(shù)據(jù)庫的數(shù)據(jù)進(jìn)行相應(yīng)處理以返回所期望的大寫字符串。

再次運(yùn)行的寫入操作現(xiàn)在將打印以下內(nèi)容：

{
  "id": 1,
  "name": "QUINTION",
  "surname": "TANG",
  "birth_year": null,
  "notes": "一個(gè)工程師",
  "register_date": "2023-09-10T15:29:50"
}

用于動(dòng)態(tài)查詢數(shù)據(jù)的列屬性

在 Python SQLAlchemy 中，column_property 函數(shù)用于在 SQLAlchemy 類上創(chuàng)建計(jì)算屬性或別名屬性，這個(gè)屬性通常源自數(shù)據(jù)庫表中的一列或多列。column_property 函數(shù)允許在 SQLAlchemy 模型上定義其他屬性，這些屬性不直接映射到數(shù)據(jù)庫中的單個(gè)列，而是從現(xiàn)有列數(shù)據(jù)通過一定的計(jì)算或派生。

在上面用到的 User 數(shù)據(jù)表中，通常存儲(chǔ)的是具體的年月日，而要獲取用戶的年齡就需要經(jīng)過一定的計(jì)算。

from sqlalchemy import Column, DateTime, Integer, String
from sqlalchemy.sql import func, extract
from sqlalchemy.orm import column_property, validates
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    # 計(jì)算獲取年齡列數(shù)據(jù)，此數(shù)據(jù)在數(shù)據(jù)庫中并非真實(shí)存在
    age = column_property(extract('year', func.now()) - birth_year)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())
    @validates('name', 'surname', 'notes')
    def normalize(self, _, value):
        if type(value) == str:
            return value.upper()
    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        dict_repr['age'] = self.age
        return json.dumps(dict_repr, indent=2)

上面代碼定義了新字段 age，它不是列類型，而是包含一個(gè)短表達(dá)式。使用來自 SQLAlchemy 的 sql 模塊的 extract 方法，由 func.now() 派生的時(shí)間戳中提取年份。以前見過這個(gè)，每當(dāng)查詢表并動(dòng)態(tài)派生時(shí)間戳?xí)r，該函數(shù)就會(huì)運(yùn)行。通過提取當(dāng)前時(shí)間戳的年份字段，減去用戶的出生年份，就可以計(jì)算出用戶的年齡。這個(gè)值不會(huì)存儲(chǔ)在數(shù)據(jù)表行中的任何位置，但會(huì)在查詢數(shù)據(jù)表的時(shí)候返回，是動(dòng)態(tài)計(jì)算的。

由于字段 age 不是實(shí)際的列，因此它的值不會(huì)包含在 self.__table__.columns 字典中。當(dāng)然還有其他的方法來達(dá)到這一目的。

下面來看看寫入數(shù)據(jù)庫和查詢數(shù)據(jù)的結(jié)果。

user_sign_up = UserSignUp(
    name="Quintion",
    surname="Tang",
    notes="一個(gè)工程師",
    birth_year=1988
)
try:
    user_to_store = User(
        name=user_sign_up.name,
        surname=user_sign_up.surname,
        birth_year=user_sign_up.birth_year,
        notes=user_sign_up.notes
    )
    db_session.add(user_to_store)
    db_session.commit()
    print(user_to_store)
except IntegrityError as e:
    db_session.rollback()
    print(f"用戶注冊(cè)時(shí)出錯(cuò)：{str(e)}")

這次，提供出生年份，以便執(zhí)行年齡計(jì)算。

{
  "id": 1,
  "name": "QUINTION",
  "surname": "TANG",
  "birth_year": 1988,
  "notes": "一個(gè)工程師",
  "register_date": "2023-09-10T15:58:31",
  "age": 35
}

現(xiàn)在，字段 age 會(huì)動(dòng)態(tài)填充并在每次查詢時(shí)返回。

主鍵之上多列的唯一約束

熟悉 SQL 數(shù)據(jù)庫系統(tǒng)的人從一開始就知道每個(gè)表都需要包含一個(gè)主鍵。主鍵是唯一標(biāo)識(shí)表中每條記錄（行）的基本概念。它確保指定列中不存在重復(fù)值或空值，從而保證數(shù)據(jù)的完整性和唯一性。

然而，有時(shí)，需要對(duì)記錄條目進(jìn)行更多限制。在某些場(chǎng)景中，主鍵并不能滿足所有需求。

舉一個(gè)例子，假設(shè)注冊(cè)用戶與現(xiàn)有數(shù)據(jù)表中的用戶具有完全相同的 name 和 surname ，在表主鍵只是一個(gè)增量數(shù)字的時(shí)候，是允許插入的。為了說明問題，驗(yàn)證相同用戶名信息是否是同一個(gè)人，可以通過使用字段 name 、 surname 和 birth_year 來標(biāo)識(shí)兩個(gè)不同的記錄，也就是說，如果現(xiàn)有用戶的所有的這三個(gè)字段都相同，則應(yīng)拒絕插入，這種規(guī)則稱為唯一性索引。

下面來看看如何在表中強(qiáng)制執(zhí)行這樣的規(guī)則：

import json
from sqlalchemy import UniqueConstraint
from sqlalchemy import Column, DateTime, Integer, String
from sqlalchemy.sql import func, extract
from sqlalchemy.orm import column_property, validates
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    age = column_property(extract('year', func.now()) - birth_year)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())
    # 增加多字段唯一性規(guī)則
    __table_args__ = (UniqueConstraint('name', 'surname', 'birth_year', name='unique_fullname_per_birth_year'),)
    @validates('name', 'surname', 'notes')
    def normalize(self, _, value):
        if type(value) == str:
            return value.upper()
    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        dict_repr['age'] = self.age
        return json.dumps(dict_repr, indent=2)

對(duì)象 UniqueConstraint 提供的表參數(shù)，對(duì)象采用不同的列名作為參數(shù)，這些列名稱形成每個(gè)記錄的唯一性，還可以為規(guī)則提供名稱。

這一特性就不再演示其效果了。下面是完整代碼：

from datetime import datetime
from pydantic import BaseModel
from typing import Optional
import os
import json
from sqlalchemy import create_engine, UniqueConstraint
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, DateTime, Integer, String
from sqlalchemy.sql import func, extract
from sqlalchemy.orm import column_property, validates
project_dir = os.path.dirname(os.path.abspath(__file__))
engine = create_engine("sqlite:///{}".format(os.path.join(project_dir, "./storage.db")),
                       connect_args={"check_same_thread": False})
Base = declarative_base()
class UserSignUp(BaseModel):
    name: str
    surname: Optional[str] = None
    birth_year: Optional[int] = None
    notes: Optional[str] = None
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=True)
    surname = Column(String, nullable=True)
    birth_year = Column(Integer, nullable=True)
    age = column_property(extract('year', func.now()) - birth_year)
    notes = Column(String, nullable=True)
    register_date = Column(DateTime, default=func.now())
    __table_args__ = (UniqueConstraint('name', 'surname', 'birth_year', name='unique_fullname_per_birth_year'),)
    @validates('name', 'surname', 'notes')
    def normalize(self, _, value):
        if type(value) == str:
            return value.upper()
    def __repr__(self):
        dict_repr = {c.name: getattr(self, c.name) for c in self.__table__.columns}
        for key, value in dict_repr.items():
            if isinstance(value, datetime):
                dict_repr[key] = datetime.isoformat(value)
        dict_repr['age'] = self.age
        return json.dumps(dict_repr, indent=2)
Base.metadata.create_all(bind=engine)
session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db_session = session_local()
user_sign_up = UserSignUp(
    name="chris",
    surname="karvouniaris",
    notes="some notes about me",
    birth_year=1992
)
try:
    user_to_store = User(
        name=user_sign_up.name,
        surname=user_sign_up.surname,
        birth_year=user_sign_up.birth_year,
        notes=user_sign_up.notes
    )
    db_session.add(user_to_store)
    db_session.commit()
    print(user_to_store)
except IntegrityError as e:
    db_session.rollback()
    print(f"用戶注冊(cè)時(shí)出錯(cuò)：{str(e)}")