接下來兩天我們來說明如何使用SVCS、FastHTML搭配EdgeDB建立一個Python todo app（註1），並希望能夠提供以下三個功能：

• 新增todo。
• 刪除todo。
• 取得當前全部todo。

今天我們會先說明SVCS的基本使用方法及編寫所需的EdgeDB schema與query，明天則實際使用FastHTML來建立一個實際可以執行的app。

SVCS簡介

不知道您有沒有遇過當想取得某種服務，像是database connnection、各種client或是快取等，總是重覆寫著類似的程式到處複製貼上，或是需要將服務作為參數傳來傳去呢？

雖然FastAPI的Depends功能，很大程度上簡化了這個步驟；但是SVCS卻讓我有了一種「隨取即用」的體驗。SVCS是由attrs與structlog的主要維護者Hynek Schlawack所開發的服務取得套件。其原理是使用svcs.Registry來註冊服務及使用svcs.Container來尋找服務並管理該服務的lifecycle（底層使用context manager）。

我必須承認SVCS的文件講解得非常清楚，其源碼也精鍊易讀，但就像薛丁格的貓一樣，我總是介於懂與不懂之間。不過還好Hynek可能理解了凡人如我的痛苦，特地針對了AIOHTTP、FastAPI、Flask、Pyramid及Starlette開發了好用的整合函數，大大降低了其使用難度。由於FastHTML底層也是使用Starlette，與SVCS可謂是絕配呀！

下面我們參考其官方文件，說明如何在Starlette中進行初始化及取得服務（註2）。

初始化

初始化需要完成兩件工作：

• 定義一個使用@svcs.starlette.lifespan()裝飾的函數（或是generator）作為fast_app的lifespan參數。在該函數內可使用svcs.Registry來註冊所需要的服務。如果其內有使用yield的話（可以選擇性給定回傳值），將作為分界：在yield之前的程式碼會在app開始後被呼叫；在yield之後的程式碼則會在app結束前被呼叫。
• 將svcs.starlette.SVCSMiddleware加入到fast_app的middleware參數。

from starlette.applications import Starlette
from starlette.middleware import Middleware

import svcs


@svcs.starlette.lifespan
async def lifespan(app: Starlette, registry: svcs.Registry):
    registry.register_factory(Database, Database.connect)

    yield {"your": "other stuff"}

    # Registry is closed automatically when the app is done.


app = Starlette(
    lifespan=lifespan,
    middleware=[Middleware(svcs.starlette.SVCSMiddleware)],
    routes=[...],
)

取得服務

在任何的view function中，只要能夠捕抓到request，即可以透過svcs.starlette.aget()來獲取所註冊的服務。

import svcs

async def view(request):
    db = await svcs.starlette.aget(request, Database)

EdgeDB schema與query

此處我們定義EdgeDB的schema及三個query。最後使用edgedb-python套件來作為query builder幫助我們將query翻譯為Python程式碼。

Schema

這個schema存在dbschema/default.esdl中，其內定義了：

• Todo object type
  內含一個名為「"title"」的property，且為required及single，即每次insert時都要提供一個str型態。此外，其使用了exclusive()來確保title不會重覆及使用了max_len_value()及min_len_value()來設定字數上下限。
• select_todo_by_id()
select_todo_by_id()功能為使用id property作為篩選todo的依據。請留意這裡同時使用了assert_exists()及assert_single()來確保返回值是存在且唯一符合條件的todo。

module default {
   type Todo {
      required title: str {
         constraint exclusive;
         constraint min_len_value(1);
         constraint max_len_value(50);
      };
   }

   function select_todo_by_id(tid: uuid) -> Todo
   using (
      select (assert_exists(assert_single((select Todo filter .id=tid))))
   )
}

當schema備妥後，我們可以使用edgedb project init進行初始化，並使用edgedb migration create及edgedb migrate來進行migration。如果對這些步驟不太熟悉的朋友，請參考[Day02]的說明。

新增Todo

此query存在app/queries/create_todo.edgeql中：

with title:= <str>$title,
     todo:= (insert Todo {title:=title})
select todo {id, title};

首先在with區塊中使用<str>來指定使用者所輸入的$title型別，並命名為title。接著insert一個Todo object type並命名為todo。

最後使用select與shape來印出此todo的id及title property。

刪除Todo

此query存在app/queries/delete_todo.edgeql中：

with id:= <uuid><str>$id,
     todo:= (select (select_todo_by_id(id)))
select (delete todo) {id, title};

在with區塊中使用<uuid><str>來指定使用者所輸入的$id型別，並命名為id。接著利用select_todo_by_id()搭配id來選擇符合條件的todo，並命名為todo。

最後使用delete todo刪除此todo，並使用select與shape來印出此todo的id及title property。

取得Todos

此query存在app/queries/get_todos.edgeql中：

select Todo {id, title};

使用select與shape來印出所有Todo object type的id及title property。

Query builder

請留意執行query builder時，該EdgeDB instance必須是running的狀態。您可以使用edgedb instance list來列出所有的instance，並使用edgedb instance start -I xxx來啟動名為xxx的instance。

打開命令列，安裝edgedb-python套件，並cd移至queries資料夾後，執行下列指令：

edgedb-py

您應該可以看到類似下面的輸出（Windows系統）：

Found EdgeDB project: ...\ithome_todo
Processing ...\ithome_todo\app\queries\create_todo.edgeql
Processing ...\ithome_todo\app\queries\delete_todo.edgeql
Processing ...\ithome_todo\app\queries\get_todos.edgeql
Generating ...\ithome_todo\app\queries\create_todo_async_edgeql.py
Generating ...\ithome_todo\app\queries\delete_todo_async_edgeql.py
Generating ...\ithome_todo\app\queries\get_todos_async_edgeql.py
Done.

edgedb-python在「看過」我們所寫的三個query後，確認了其語法是正確的EdgeQL後，會產生三個相對應的Python檔案。舉例來說，get_todos_async_edgeql.py將get_todos.edgeql的query翻譯為下面的Python程式碼：

# AUTOGENERATED FROM 'app/queries/get_todos.edgeql' WITH:
#     $ edgedb-py


from __future__ import annotations
import dataclasses
import edgedb
import uuid


class NoPydanticValidation:
    @classmethod
    def __get_pydantic_core_schema__(cls, _source_type, _handler):
        # Pydantic 2.x
        from pydantic_core.core_schema import any_schema
        return any_schema()

    @classmethod
    def __get_validators__(cls):
        # Pydantic 1.x
        from pydantic.dataclasses import dataclass as pydantic_dataclass
        pydantic_dataclass(cls)
        cls.__pydantic_model__.__get_validators__ = lambda: []
        return []


@dataclasses.dataclass
class GetTodosResult(NoPydanticValidation):
    id: uuid.UUID
    title: str


async def get_todos(
    executor: edgedb.AsyncIOExecutor,
) -> list[GetTodosResult]:
    return await executor.query(
        """\
        select Todo {id, title};\
        """,
    )

雖然看似有一點複雜，但對使用者而言，我們只需要import最終可以執行query的函數，如get_todos()即可。

使用query builder不但可以保證query的正確性，還可以省去手動翻譯至Python的麻煩步驟，推薦給大家使用。

備註

註1：參考自FastHTML Gallery的Todo list範例。

註2：或許您也會有興趣一併學習SVCS的服務健康檢查功能。

程式碼傳送門

本日所有程式碼可參考fasthtml-svcs-edgedb-mvp repo。