SQLAlchemy (二) — 如何使用 SQLAlchemy ORM

繼上一篇 SQLAlchemy 簡介後，這次要來實作精簡版的 SQLAlchemy 程式範例。

實作流程如下：

創建 Engine

定義 ORM Model

在目標 Database 生成資料表

使用 Session 操作資料

1. 創建 Engine

首先創建 Engine，用來建立 DB 連線

Engine 是什麼？

• Engine 是連接到特定資料庫的中心來源。它提供了一個工廠和稱為連接池（Connection Pool）的儲存空間，用於管理資料庫連接。
• Engine 通常作為全局物件存在。也就是說，針對特定的資料庫伺服器只會創建一次。

from sqlalchemy import create_engine
engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)

create_engine 主要參數為一個 URL sqlite+pysqlite:///:memory:，包含了三個重點：

• 我們正在與哪種資料庫溝通？
  在這個例子中為 sqlite
• 我們使用哪個 DBAPI？
  Python 的 DBAPI 是 SQLAlchemy 用來與特定數據庫互動的第三方驅動。在這個例子中，我們使用的是 pysqlite，它在現代 Python 中是 SQLite 的 sqlite3 標準庫接口。
• 我們如何定位資料庫？
  在這個例子中為 /:memory，這是對 sqlite3 模組的指示，表明我們將使用一個僅存在於記憶體中的資料庫。這種類型的數據庫非常適合實驗，因為它不需要任何服務器，也不需要創建新檔案。

2. 宣告 ORM Model

在 SQLAlchemy 中宣告模型（Model），建立 Python 類別與資料庫 Table 之間的映射關係。

from sqlalchemy import Column, Intege, Text
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(Text)
    content = Column(Text)
    view_count = Column(Integer, default=0)

• 透過 Model，你可以在 Python 定義資料庫的結構（例如，表、欄位、索引、關聯、約束等）。
• 這些用來描述資料庫結構的資訊，被稱作 DataBase MetaData。

3. 在目標 Database 創建資料表

使用 Model 定義的資料結構和 Engine，就能針對目標 Database 生成 Table

Base.metadata.create_all(engine)

當你呼叫 metaData.create_all()，SQLAlchemy 會查看 metadata 中定義的所有 Table Object，並生成 SQL 語句 CREATE TABLE ，在資料庫中創建 Table。

例如以上 Model 定義後，呼叫 create_all() 會生成 SQL 如下：

BEGIN
PRAGMA main.table_info("article")
PRAGMA temp.table_info("article")

CREATE TABLE article (
        id INTEGER NOT NULL, 
        title TEXT, 
        content TEXT, 
        view_count INTEGER, 
        PRIMARY KEY (id)
)

COMMIT

資料表的創建，通常會在應用程式啟動初期進行，確保必要的資料庫架構已到位。

如果某個 Table 在資料庫已存在，SQLAlchemy 不會再重新創建它。（default 設定）

4. 使用 Session 操作資料

使用 Session 和 ORM 物件，就能實現資料的 CRUD

創建資料

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

with Session() as session:
    welcome_article = Article(title='Hello World', content='This is my first article')
    announcement_article = Article(title='Important Announcement', content='This is an important announcement')
    session.add_all([welcome_article, announcement_article])
    session.commit()

• sessionmaker是一個工廠函數，呼叫時會產生新的 Session 物件。
• 一個 Session 代表與資料庫的一次 “對話”，允許您查詢和操作數據。

更新資料

with Session() as session:
    article = session.query(Article).filter(Article.id == article_id).first()
    article.view_count += 1
    session.commit()

查詢

with Session() as session:
    article = session.query(Article).filter(Article.id == article_id).first()

    print(f"title: {article.title}, view_count: {article.view_count}")
    return article

# title: Hello World, view_count: 1

以上就是一個精簡版的 SQLAlchemy 使用範例

最後附上完整程式碼

from sqlalchemy import Column, Integer, Text
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase


class Base(DeclarativeBase):
    pass


class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(Text)
    content = Column(Text)
    view_count = Column(Integer, default=0)


engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)

Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)


def insert_article():
    with Session() as session:
        welcome_article = Article(title='Hello World', content='This is my first article')
        announcement_article = Article(title='Important Announcement', content='This is an important announcement')

        session.add_all([welcome_article, announcement_article])
        session.commit()


def update_article_count(article_id: int):
    with Session() as session:
        article = session.query(Article).filter(Article.id == article_id).first()
        article.view_count += 1
        session.commit()


def get_article(article_id: int):
    with Session() as session:
        article = session.query(Article).filter(Article.id == article_id).first()

        print(f"title: {article.title}, view_count: {article.view_count}")
        return article


def main():
    insert_article()
    update_article_count(article_id=1)
    get_article(article_id=1)


main()