前兩天都只有建立資料庫，今天開始來操作資料庫~

版本

在開始之前，先說一下，在今年 1 月，SQLAlchemy 正式發布了 2.0.0 版 (公告連結)，語法上有不少地方有調整，也有新的操作方式 (Query 那部份)，但因為新語法我還不熟悉，再加上大部分 1.4.X 的語法仍然支援，因此以下的操作都是 1.4.X 的「舊」語法。

再補充一個題外話，這次更新讓我看到了我所見過的最長的 Migration guide...

真的很長，可以進去看一眼體會一下 XD

CRUD

操作資料庫的動作基本上有四種：新增 (create)、刪除 (delete)、查詢 (read，意思有一點點不同)、修改 (update)，中文簡稱為「增刪查改」，英文則是「CRUD」

SQLAlchemy 的 CRUD

讓我們先來看一個查詢的範例

def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

應該很好理解，它就是從資料庫連線開始，先用 query 指定是哪個 Table (用 models 內的 class)，再用 filter 根據條件過濾出要操作的對象，最後再用 first 指定第一筆。

除了 first() 之外，常用的還有 all()、one()、one_or_none

如果要再添加其他條件，就直接一路接下去就好 (或是插在中間)，用起來很方便。

如果大家以前有學過 SQL，就會發現這樣的寫法真的很不一樣 XD

刪除或是修改是用 delete() 和 update()，而新增的話則是用 add()，但前面就不用 query() 了。

除了查詢之外，其他三種都有對資料庫做修改，因此最後都要再多補一行 db.commit() 才會生效。

與 FastAPI 整合

上面講了很多怎麼做，這邊直接來看範例比較快

# crud.py
from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

上面這邊有查尋和新增，相信大家應該可以看出寫法了 XD

裡面有使用到 refresh()，這部份以後有機會再仔細討論，這邊只要先知道它就是要更新一下資料而已 (因為建立連線的當下還沒有這筆資料，是 commit 過後才有的)

看一下 API 的部份範例 (完整的請看這邊)

# main.py
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

接著用 Postman 測試

可以看到我們成功新增並且後端有把我們新增的資料回傳到前端了

接著，再手動打開 test.db 確認

可以看到真的有新增到資料庫內，並且內容與剛剛收到的 Response 是相符的

直接下指令

最後補充另一個 SQLAlchemy 支援的寫法，那就是用 text() 和 execute() 來用 SQL 操作 (文件)

一樣，我相信看過範例就懂了~

from sqlalchemy import text

t = text("SELECT * FROM users")
result = connection.execute(t)

因此如果不習慣 ORM 的語法的朋友，還是可以先用 SQL，只是要多小心一下 SQL Injection 之類的問題。

重點回顧

今天寫的比較趕，只能快速地介紹 CRUD 而已...

明天會開始介紹資料庫的 migration