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

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

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

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

本文一起來學習 SQLAlchemy 庫的三個有用功能，在大部分的項目開發(fā)中都可能用得上的功能。

下面先來了解一些基礎知識。

基礎知識

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

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

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

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

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

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

存儲 Storage

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

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)

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

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

表用戶行被定義為包含比我們?yōu)橛脩糇砸氲臄?shù)據(jù)模型額外的兩個字段。

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

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

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

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

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

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

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)建一個要注冊的新用戶。

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

最后，使用 Users 的數(shù)據(jù)庫表類創(chuàng)建一個條目，以使用建立的會話寫入數(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"用戶注冊時出錯：{str(e)}")

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

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

請注意，birth_year 字段為空，因為首先沒有為其提供任何整數(shù)，并且在寫入操作期間會自動為 register_date 生成時間戳。

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

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

來看看將如何在這里做到這一點。例如，假設想要將每個字符串字段大寫。如果需要在數(shù)據(jù)庫中對數(shù)據(jù)進行標準化，這樣的過程將對于后期數(shù)據(jù)處理分析非常有用。統(tǒng)一字符串字段大寫的好處是在后續(xù)相關(guā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())
    # 這里對數(shù)據(jù)進行預處理
    @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)

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

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

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

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

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

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

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

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)
    # 計算獲取年齡列數(shù)據(jù)，此數(shù)據(jù)在數(shù)據(jù)庫中并非真實存在
    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，它不是列類型，而是包含一個短表達式。使用來自 SQLAlchemy 的 sql 模塊的 extract 方法，由 func.now() 派生的時間戳中提取年份。以前見過這個，每當查詢表并動態(tài)派生時間戳時，該函數(shù)就會運行。通過提取當前時間戳的年份字段，減去用戶的出生年份，就可以計算出用戶的年齡。這個值不會存儲在數(shù)據(jù)表行中的任何位置，但會在查詢數(shù)據(jù)表的時候返回，是動態(tài)計算的。

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

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

user_sign_up = UserSignUp(
    name="Quintion",
    surname="Tang",
    notes="一個工程師",
    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"用戶注冊時出錯：{str(e)}")

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

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

現(xiàn)在，字段 age 會動態(tài)填充并在每次查詢時返回。

主鍵之上多列的唯一約束

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

然而，有時，需要對記錄條目進行更多限制。在某些場景中，主鍵并不能滿足所有需求。

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

下面來看看如何在表中強制執(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)

對象 UniqueConstraint 提供的表參數(shù)，對象采用不同的列名作為參數(shù)，這些列名稱形成每個記錄的唯一性，還可以為規(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"用戶注冊時出錯：{str(e)}")