今天會講解如何在 FastAPI 中與其他的 NoSQL 資料庫做連結。這次會使用 Amazon DynamoDB 作為 DB 來進行操作，我們會使用 aws 官方的 boto3 套件及 local 的 DynamoDB 進行模擬。

安裝 Local DynamoDB

我們可以使用 Docker 來快速開啟一個本地的容器，首先我們先把映像檔 PULL 到本機上：

docker pull amazon/dynamodb-local

接著就可以使用 docker run 來啟動一個容器：

docker run -d -p 8000:8000 amazon/dynamodb-local

啟動後可以看到：

容器內已經初始化好你的 database。

安裝 DynamoDB admin

DynamoDB admin 是一個 GUI 可以讓你管理連接到的 DynamoDB，我們可以透過使用 npm 來進行安裝：

npm install -g dynamodb-admin

安裝好後我們只需要在機器的環境變數上設置 DYNAMO_ENDPOINT 這個變數使其可以連接到你的 Database 即可使用：

# Powershell
$ENV:DYNAMO_ENDPOINT='http://localhost:8000'

# Windows CMD
set DYNAMO_ENDPOINT=http://localhost:8000

# Mac/Linux
DYNAMO_ENDPOINT=http://localhost:8000

設置好之後只要在終端機上輸入

dynamodb-admin

就可以開啟你的 DynamoDB Admin：

新增資料及拿取資料

我們可以開始從 FastAPI 中將資料上傳到 DynamoDB 中了，但是上傳前我們需要先新增 table 以儲放我們的資料，這一步驟可以使用 boto3 程式碼執行，也可以使用 admin 執行即可。

這裡我們使用 admin 直接新增一張 table 即可：

設定資源連線

新增完表後，我們就可以來設定程式的部分了，我們首先要先能建立一個 boto3 的 ServiceResource 可以連線到 Database：

# database.py

from boto3.resources.base import ServiceResource
from boto3 import resource

dynamodb: ServiceResource = resource(
    "dynamodb",
    endpoint_url='http://localhost:8000',
    region_name='us-east-1',  # note that if you create a table using different region name and aws key
    aws_access_key_id='key',  # you won't see this table on the admin app
    aws_secret_access_key='key',
) 

因為是使用 local 的 database 所以預設通常會是以上的設定，如果有需要遷移到雲端時只需要修改這裡的變數即可。

定義輸入

# schemas.py

from pydantic import BaseModel
from typing import Optional

class PutDataInput(BaseModel):
    name: Optional[str]
    mail: Optional[str]

簡單給兩個輸入進行測試即可。

撰寫業務邏輯

from botocore.exceptions import ClientError
from src.database import dynamodb
from src import schemas

table_name = "test_table"

def get_data(id: str):
    """
    Get DynamoDB data
    """
    table = dynamodb.Table(table_name)
    try:
        response = table.get_item(Key={'id': id})
    except ClientError as e:
        print(e)

    try:
        data = response['Item']
    except KeyError as e:
        print(e)
        data = None
    return data

def put_data(id: str, data: schemas.PutDataInput):
    """
    Put DynamoDB data
    """

    data = data.dict()
    table = dynamodb.Table(table_name)
    try:
        table.put_item(Item=data)
    except ClientError as e:
        print(e)
    return data

我們透過定義資料表名稱以存取剛剛建立的資料表，透過 get_item 和 put_item 來存取或上傳資料，這邊有個小細節，boto3 並不支援非同步，所以定義函式時聲明為同步的函式即可。

路徑操作函式

@app.get("/test/{id}")
def get_data(id: str):
    return service.get_data(id)

@app.put("/test/")
def put_data(id: str, data: schemas.PutDataInput):
    return service.put_data(id, data)

就如同剛剛所說的我們的業務邏輯函式不是非同步的，所以說在路徑操作函式這裡宣告為同步的函式即可，FastAPI 會自動幫我們放到 theading pool 中。

成功結果

我們試著上傳一筆測試資料。

再使用 get 去拿取剛剛新增的資料，可以獲得資料代表上傳成功。

你也可以在 admin 中看到剛剛新增那筆資料已經進去資料庫了。

小結

今天使用 DynamoDB 作為範例跟大家說該怎麼去串接一個 NoSQL 資料庫，用其他的資料庫來做也是大同小異，要注意的是如果函式不支援非同步的話請用同步就好，FastAPI 還是會幫你做一個基礎的效能優化的。

參考資料

Introduction to FastAPI and Local DynamoDB | by Agus Richard | Nerd For Tech | Medium