首頁 >後端開發 >Python教學 >使用 Flama 進行本機域驅動設計

使用 Flama 進行本機域驅動設計

Barbara Streisand原創: 2024-11-03 14:08:03691瀏覽

您可能已經聽說過最近發布的 Flama 1.7，它帶來了一些令人興奮的新功能，可以幫助您開發和生產 ML API。這篇文章專門討論此版本的主要亮點之一：對領域驅動設計的支援。但是，在我們透過實際範例深入了解細節之前，我們建議您記住以下資源（如果您還沒有熟悉它們，請先熟悉一下）：

Flama 官方文件：Flama 文檔
介紹 Flama for ML API 的貼文：用於穩健機器學習 API 的 Flama 簡介

現在，讓我們開始使用新功能，看看如何利用它來建立強大且可維護的 ML API。

什麼是領域驅動設計？

簡要概述

在現代軟體開發中，將業務邏輯與應用程式的技術設計保持一致至關重要。這就是領域驅動設計 (DDD) 的閃光點。 DDD 強調建立反映業務核心領域的軟體，透過圍繞業務概念組織程式碼來分解複雜的問題。透過這樣做，DDD 可協助開發人員創建可維護、可擴展且健壯的應用程式。以下我們將介紹您應該了解的 DDD 中最重要的概念。在我們深入探討它們之前，我們需要指出的是，這篇文章並不是要成為 DDD 的綜合指南，也不是該主題的主要參考文獻的替代品。事實上，我們推薦以下資源來更深入了解 DDD：

Harry Percival 和 Bob Gregory 的 Cosmic Python：這本書是學習如何在 Python 中應用 DDD 的絕佳資源。
Eric Evans 的《領域驅動設計：解決軟體核心的複雜性》：這本書向世界介紹了 DDD，對於任何有興趣深入了解 DDD 的人來說，這是一本必讀的書。

關鍵概念

在深入研究DDD 的任何關鍵概念之前，我們建議您看一下Cosmic Python 的一個非常有用的圖，其中這些圖顯示在應用程式的上下文中，從而顯示它們是如何互連的：圖.

領域模型

領域模型的概念可以透過其術語的簡單定義來解釋：

領域是指我們的軟體所支援的活動（或知識）的特定主題領域。
模型是指我們嘗試在軟體中編碼的系統或流程的簡單表示（或抽象）。

因此，域模型是一種奇特（但標準且有用）的方式來引用企業主心中關於業務如何運作的一組概念和規則。這也是我們通常所說的應用程式的業務邏輯，包括控制系統行為的規則、約束和關係。

從現在開始，我們將把域模型稱為模型。

儲存庫模式

儲存庫模式是一種設計模式，允許將模型與資料存取解耦。儲存庫模式背後的主要想法是在應用程式的資料存取邏輯和業務邏輯之間創建一個抽象層。這個抽象層允許關注點分離，使程式碼更易於維護和測試。

在實作儲存庫模式時，我們通常定義一個介面來指定任何其他儲存庫必須實現的標準方法（AbstractRepository）。然後，使用這些方法的具體實作來定義一個特定的儲存庫，其中實作了資料存取邏輯（例如，SQLAlchemyRepository）。這種設計模式旨在隔離資料操作方法，以便它們可以在應用程式的其他地方無縫使用，例如在我們的領域模型中。

工作單元模式

工作單元模式是最終將模型與資料存取分離的缺失部分。工作單元封裝了資料存取邏輯，並提供了一種將必須在單一事務中對資料來源執行的所有操作進行分組的方法。此模式確保所有操作都以原子方式執行。

在實作工作單元模式時，我們通常會定義一個介面來指定任何其他工作單元必須實現的標準方法（AbstractUnitOfWork）。然後，使用這些方法的具體實作來定義特定的工作單元，其中實作了資料存取邏輯（例如，SQLAlchemyUnitOfWork）。這種設計允許系統地處理與資料來源的連接，而不需要更改應用程式業務邏輯的實作。

使用 Flama 實施 DDD

在快速介紹了 DDD 的主要概念之後，我們準備好深入研究使用 Flama 實現 DDD。在本節中，我們將引導您完成設定開發環境、建立基礎應用程式以及使用 Flama 實現 DDD 概念的過程。

在繼續範例之前，請先看一下 Flama 關於我們剛剛回顧的主要 DDD 概念的命名約定：

Native Domain-Driven Design with Flama

如上圖所示，命名約定非常直觀：Repository指的是儲存庫模式；並且，Worker是指工作單元。現在，我們可以繼續使用 DDD 實作 Flama API。但是，在我們開始之前，如果您需要回顧如何使用 flama 建立簡單 API 的基礎知識，或在程式碼準備好後如何執行 API，那麼您可能需要檢查出快速入門指南。在那裡，您將找到完成本文所需的基本概念和步驟。現在，事不宜遲，讓我們開始實施吧。

設定開發環境

我們的第一步是建立開發環境，並安裝該專案所需的所有依賴項。好處是，對於這個範例，我們只需要安裝 flama 即可擁有實作 JWT 驗證所需的所有工具。我們將使用詩歌來管理我們的依賴項，但如果您願意，您也可以使用 pip：

poetry add "flama[full]" "aiosqlite"

aiosqlite 套件需要將 SQLite 與 SQLAlchemy 一起使用，這是我們將在本範例中使用的資料庫。

如果你想知道我們通常如何組織我們的項目，請查看我們之前的文章，其中我們詳細解釋瞭如何使用詩歌建立 python 項目，以及我們通常遵循的項目文件夾結構。

基礎應用

讓我們從一個具有單一公共端點的簡單應用程式開始。該端點將傳回 API 的簡要描述。

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

如果你想運行這個應用程序，你可以將上面的代碼保存在src 資料夾下一個名為app.py 的文件中，然後運行以下命令（記住要激活詩歌環境，否則你需要在命令前面加上詩歌運行）：

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

其中 --server-reload 標誌是可選的，用於在程式碼變更時自動重新載入伺服器。這在開發過程中非常有用，但如果不需要，可以將其刪除。有關可用選項的完整列表，您可以執行 flama run --help，或檢視文件。

或者，您也可以透過執行下列腳本來執行應用程序，您可以將其儲存為 src 資料夾下的 __main__.py：

# src/__main__.py
import flama

def main():
    flama.run(
      flama_app="src.app:app", 
      server_host="0.0.0.0", 
      server_port=8000, 
      server_reload=True
    )

if __name__ == "__main__":
    main()

然後，您可以透過執行以下命令來執行應用程式：

poetry add "flama[full]" "aiosqlite"

DDD 在行動

現在，為我們的應用程式設定了一個最小的框架，我們可以開始實現我們剛剛在
中回顧的 DDD 概念一個試圖模仿現實世界場景的簡單範例的上下文。假設我們需要開發一個 API 來管理用戶，而我們有以下要求：

我們希望透過向 /user/ 發出 POST 請求來建立新用戶，並提供用戶的姓名、電子郵件和密碼。
建立的任何使用者都將儲存在具有以下架構的資料庫中：
- id：使用者的唯一識別碼。
- name: 使用者名稱。
- 姓氏：使用者的姓氏。
- 電子郵件：使用者的電子郵件。
- 密碼：用戶的密碼。在將其儲存到資料庫之前應該對其進行哈希處理。
- active：一個布林標誌，指示使用者是否處於活動狀態。預設情況下，用戶被建立為非活動用戶。
建立的使用者必須透過使用其電子郵件和密碼向 /user/activate/ 發送 POST 請求來啟動其帳戶。一旦用戶被激活，用戶的狀態必須在資料庫中更新為激活。
使用者可以透過使用電子郵件和密碼向 /user/signin/ 發送 POST 請求來登入。如果使用者處於活動狀態，API 必須傳回所有使用者資訊。否則，API 必須傳回錯誤訊息。
想要停用其帳戶的使用者可以透過使用其電子郵件和密碼向 /user/deactivate/ 發送 POST 請求來實現。一旦使用者被停用，使用者的狀態必須在資料庫中更新為非活動。

這組需求構成了我們先前所說的應用程式的領域模型，它本質上只不過是以下使用者工作流程的具體化：

透過向 /user/ 發送 POST 請求來建立使用者。
使用者透過向 /user/activate/ 發出 POST 請求來啟動其帳戶。
使用者透過 /user/signin/ 的 POST 請求登入。
使用者透過向 /user/deactivate/ 發出 POST 請求來停用其帳號。
使用者可以根據需要重複步驟 2-4多次。

現在，讓我們使用儲存庫和工作模式來實作領域模型。我們將從定義資料模型開始，然後實作儲存庫和工作模式。

資料模型

我們的使用者資料將儲存在 SQLite 資料庫中（您可以使用 SQLAlchemy 支援的任何其他資料庫）。我們將使用以下資料模型來表示使用者（您可以將此程式碼保存在 src 資料夾下名為 models.py 的檔案中）：

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

除了資料模型之外，我們還需要一個遷移腳本來建立資料庫和表格。為此，我們可以將以下程式碼保存在專案根目錄下名為migrations.py的檔案中：

poetry add "flama[full]" "aiosqlite"

然後，我們可以透過執行以下命令來執行遷移腳本：

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

儲存庫

在此範例中，我們只需要一個儲存庫，即處理使用者表上的原子操作的儲存庫，其名稱為 UserRepository。值得慶幸的是，flama 為與 SQLAlchemy 表相關的儲存庫提供了一個基類，稱為 SQLAlchemyTableRepository。

類別 SQLAlchemyTableRepository 提供了一組對錶執行 CRUD 操作的方法，具體為：

create：在表格中建立新元素。如果元素已經存在，則會引發異常（IntegrityError），否則將傳回新元素的主鍵。
檢索：從表格中檢索元素。如果該元素不存在，它將引發異常（NotFoundError），否則將傳回該元素。如果找到多個元素，則會引發異常（MultipleRecordsError）。
update：更新表格中的元素。如果該元素不存在，它將引發異常（NotFoundError），否則將傳回更新後的元素。
刪除：從表中刪除一個元素。
list：列出表格中與傳遞的子句和篩選器相符的所有元素。如果沒有給予子句或過濾器，它將傳回表中的所有元素。如果沒有找到元素，則傳回一個空列表。
drop：從資料庫中刪除表格。

就我們的範例而言，我們不需要對表進行任何進一步的操作，因此 SQLAlchemyTableRepository 提供的方法就足夠了。我們可以將以下程式碼保存在 src 資料夾下名為 repositories.py 的檔案中：

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

如您所見，UserRepository 類別是 SQLAlchemyTableRepository 的子類，它只需要在 _table 屬性中設定表格。這是我們為使用者表建立一個功能齊全的儲存庫所需要做的唯一事情。

如果我們想要新增標準 CRUD 操作之外的自訂方法，我們可以透過在 UserRepository 類別中定義它們來實現。例如，如果我們想要新增一個方法來統計活躍使用者數，我們可以這樣做：

# src/__main__.py
import flama

def main():
    flama.run(
      flama_app="src.app:app", 
      server_host="0.0.0.0", 
      server_port=8000, 
      server_reload=True
    )

if __name__ == "__main__":
    main()

雖然我們不會在範例中使用此方法，但很高興知道我們可以根據需要向儲存庫添加自訂方法以及它們的實作方式
在儲存庫模式的上下文中。正如我們已經看到的，這是一個強大的設計模式，因為我們可以在這裡實現所有資料存取邏輯，而無需更改應用程式的業務邏輯（在相應的資源方法中實現）。

勞工

工作單元模式用於封裝資料存取邏輯，並提供一種將必須在單一交易中對資料來源執行的所有操作進行分組的方法。在 flama 中，UoW 模式是使用 Worker 的名稱實現的。與儲存庫模式相同，flama 為與 SQLAlchemy 表相關的工作人員提供了一個基類，稱為 SQLAlchemyWorker。本質上，SQLAlchemyWorker 提供了到資料庫的連接和事務，並使用工作連接實例化其所有儲存庫。在此範例中，我們的工作人員將僅使用單一儲存庫（即 UserRepository），但如果需要，我們可以新增更多儲存庫。

我們的worker將被稱為RegisterWorker，我們可以將以下程式碼保存在src資料夾下名為workers.py的檔案中：

poetry add "flama[full]" "aiosqlite"

因此，如果我們有更多儲存庫可供使用，例如 ProductRepository 和 OrderRepository，我們可以將它們新增至工作執行緒中，如下所示：

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

就這麼簡單，我們在應用程式中實作了儲存庫和工作模式。現在，我們可以繼續實作資源方法，這些方法將提供與使用者資料互動所需的 API 端點。

資源

資源是 flama 應用程式的主要構建塊之一。它們用於表示應用程式資源（在 RESTful 資源的意義上）並定義與它們互動的 API 端點。

在我們的範例中，我們將為使用者定義一個名為 UserResource 的資源，其中包含建立、啟用、登入和停用使用者的方法。資源至少需要從 flama 內建 Resource 類別派生，儘管 flama 提供了更複雜的類別來使用，例如 RESTResource 和 CRUDResource。

我們可以將以下程式碼保存在 src 資料夾下名為 resources.py 的檔案中：

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

使用 DDD 的基礎應用程式

現在我們已經實作了資料模型、儲存庫和工作模式以及資源方法，我們需要修改先前介紹的基礎應用程序，以便一切按預期運行。我們需要：

將 SQLAlchemy 連接新增至應用程序，這是透過將 SQLAlchemyModule 作為模組添加到應用程式建構函數來實現的。
將 Worker 添加到應用程式中，這是透過將 RegisterWorker 作為元件添加到應用程式建構函數中來實現的。

這將使 app.py 檔案如下：

poetry add "flama[full]" "aiosqlite"

您應該已經很清楚 DDD 模式如何讓我們能夠將應用程式的業務邏輯（在資源方法中很容易閱讀）與資料存取邏輯分開。 🎜>（在儲存庫和工作模式中實作）。另外值得注意的是，這種關注點分離如何使程式碼更易於維護和測試，以及程式碼現在如何更符合我們在本範例開始時給出的業務需求。

運行應用程式

在執行任何命令之前，請檢查您的開發環境是否設定正確，且資料夾結構如下：

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

如果一切設定正確，您可以透過執行以下命令來執行應用程式（請記住在執行應用程式之前執行遷移腳本）：

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

現在我們可以嘗試剛剛實現的業務邏輯。請記住，您可以使用curl或Postman等工具來嘗試此操作，也可以透過在瀏覽器中導航到http://localhost:8000/docs/來使用flama提供的自動產生的文檔UI並從那裡嘗試端點。

Native Domain-Driven Design with Flama

創建用戶

要建立用戶，您可以使用以下有效負載向 /user/ 發送 POST 請求：

# src/__main__.py
import flama

def main():
    flama.run(
      flama_app="src.app:app", 
      server_host="0.0.0.0", 
      server_port=8000, 
      server_reload=True
    )

if __name__ == "__main__":
    main()

因此，我們可以使用curl來發送請求，如下所示：

poetry run python src/__main__.py

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

如果請求成功，您應該收到正文為空的 200 回應，並且將在資料庫中建立使用者。

# src/models.py
import uuid

import sqlalchemy
from flama.sqlalchemy import metadata
from sqlalchemy.dialects.postgresql import UUID

__all__ = ["user_table", "metadata"]

user_table = sqlalchemy.Table(
    "user",
    metadata,
    sqlalchemy.Column("id", UUID(as_uuid=True), primary_key=True, nullable=False, default=uuid.uuid4),
    sqlalchemy.Column("name", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("surname", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("email", sqlalchemy.String, nullable=False, unique=True),
    sqlalchemy.Column("password", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("active", sqlalchemy.Boolean, nullable=False),
)

因此，我們可以使用curl來發送請求，如下所示：

# migrations.py
from sqlalchemy import create_engine

from src.models import metadata

if __name__ == "__main__":
    # Set up the SQLite database
    engine = create_engine("sqlite:///models.db", echo=False)

    # Create the database tables
    metadata.create_all(engine)

    # Print a success message
    print("Database and User table created successfully.")

鑑於使用者不活躍，您應該收到類似以下回應：

> poetry run python migrations.py

Database and User table created successfully.

我們也可以測試如果有人嘗試使用錯誤密碼登入會發生什麼：

# src/repositories.py
from flama.ddd import SQLAlchemyTableRepository

from src import models

__all__ = ["UserRepository"]

class UserRepository(SQLAlchemyTableRepository):
    _table = models.user_table

在這種情況下，您應該收到包含以下正文的 401 回應：

# src/repositories.py
from flama.ddd import SQLAlchemyTableRepository

from src import models

__all__ = ["UserRepository"]

class UserRepository(SQLAlchemyTableRepository):
    _table = models.user_table

    async def count_active_users(self):
        return len((await self._connection.execute(self._table.select().where(self._table.c.active == True))).all())

最後，我們也應該嘗試使用不存在的使用者登入：

# src/workers.py
from flama.ddd import SQLAlchemyWorker

from src import repositories

__all__ = ["RegisterWorker"]


class RegisterWorker(SQLAlchemyWorker):
    user: repositories.UserRepository

在這種情況下，您應該收到包含以下正文的 404 回應：

# src/workers.py
from flama.ddd import SQLAlchemyWorker

from src import repositories

__all__ = ["RegisterWorker"]

class RegisterWorker(SQLAlchemyWorker):
    user: repositories.UserRepository
    product: repositories.ProductRepository
    order: repositories.OrderRepository

用戶啟用設定

探索了登入程序後，我們現在可以透過使用使用者的憑證向 /user/activate/ 發送 POST 請求來啟動使用者：

# src/resources.py
import hashlib
import http
import uuid

from flama import types
from flama.ddd.exceptions import NotFoundError
from flama.exceptions import HTTPException
from flama.http import APIResponse
from flama.resources import Resource, resource_method

from src import models, schemas, worker

__all__ = ["AdminResource", "UserResource"]

ENCRYPTION_SALT = uuid.uuid4().hex
ENCRYPTION_PEPER = uuid.uuid4().hex

class Password:
    def __init__(self, password: str):
        self._password = password

    def encrypt(self):
        return hashlib.sha512(
            (hashlib.sha512((self._password + ENCRYPTION_SALT).encode()).hexdigest() + ENCRYPTION_PEPER).encode()
        ).hexdigest()

class UserResource(Resource):
    name = "user"
    verbose_name = "User"

    @resource_method("/", methods=["POST"], name="create")
    async def create(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserDetails]):
        """
        tags:
            - User
        summary:
            User create
        description:
            Create a user
        responses:
            200:
                description:
                    User created in successfully.
        """
        async with worker:
            try:
                await worker.user.retrieve(email=data["email"])
            except NotFoundError:
                await worker.user.create({**data, "password": Password(data["password"]).encrypt(), "active": False})

        return APIResponse(status_code=http.HTTPStatus.OK)

    @resource_method("/signin/", methods=["POST"], name="signin")
    async def signin(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserCredentials]):
        """
        tags:
            - User
        summary:
            User sign in
        description:
            Create a user
        responses:
            200:
                description:
                    User signed in successfully.
            401:
                description:
                    User not active.
            404:
                description:
                    User not found.
        """
        async with worker:
            password = Password(data["password"])
            try:
                user = await worker.user.retrieve(email=data["email"])
            except NotFoundError:
                raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND)

            if user["password"] != password.encrypt():
                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED)

            if not user["active"]:
                raise HTTPException(
                    status_code=http.HTTPStatus.BAD_REQUEST, detail=f"User must be activated via /user/activate/"
                )

        return APIResponse(status_code=http.HTTPStatus.OK, schema=types.Schema[schemas.User], content=user)

    @resource_method("/activate/", methods=["POST"], name="activate")
    async def activate(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserCredentials]):
        """
        tags:
            - User
        summary:
            User activate
        description:
            Activate an existing user
        responses:
            200:
                description:
                    User activated successfully.
            401:
                description:
                    User activation failed due to invalid credentials.
            404:
                description:
                    User not found.
        """
        async with worker:
            try:
                user = await worker.user.retrieve(email=data["email"])
            except NotFoundError:
                raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND)

            if user["password"] != Password(data["password"]).encrypt():
                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED)

            if not user["active"]:
                await worker.user.update({**user, "active": True}, id=user["id"])

        return APIResponse(status_code=http.HTTPStatus.OK)

    @resource_method("/deactivate/", methods=["POST"], name="deactivate")
    async def deactivate(self, worker: worker.RegisterWorker, data: types.Schema[schemas.UserCredentials]):
        """
        tags:
            - User
        summary:
            User deactivate
        description:
            Deactivate an existing user
        responses:
            200:
                description:
                    User deactivated successfully.
            401:
                description:
                    User deactivation failed due to invalid credentials.
            404:
                description:
                    User not found.
        """
        async with worker:
            try:
                user = await worker.user.retrieve(email=data["email"])
            except NotFoundError:
                raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND)

            if user["password"] != Password(data["password"]).encrypt():
                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED)

            if user["active"]:
                await worker.user.update({**user, "active": False}, id=user["id"])

        return APIResponse(status_code=http.HTTPStatus.OK)

透過此請求，使用者應該被激活，並且您應該收到一個空正文的 200 回應。

與前面的情況一樣，我們也可以測試如果有人嘗試使用錯誤的密碼來啟動使用者會發生什麼：

poetry add "flama[full]" "aiosqlite"

在這種情況下，您應該收到包含以下正文的 401 回應：

# src/app.py
from flama import Flama

app = Flama(
    title="Domain-driven API",
    version="1.0.0",
    description="Domain-driven design with Flama ?",
    docs="/docs/",
)

@app.get("/", name="info")
def info():
    """
    tags:
        - Info 
    summary:
        Ping
    description:
        Returns a brief description of the API
    responses:
        200:
            description:
                Successful ping.
    """
    return {"title": app.schema.title, "description": app.schema.description, "public": True}

最後，我們也應該嘗試啟動一個不存在的使用者：

flama run --server-reload src.app:app

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

在這種情況下，您應該收到包含以下正文的 404 回應：

# src/__main__.py
import flama

def main():
    flama.run(
      flama_app="src.app:app", 
      server_host="0.0.0.0", 
      server_port=8000, 
      server_reload=True
    )

if __name__ == "__main__":
    main()

用戶啟動後登入

現在用戶已激活，我們可以嘗試再次登入：

poetry run python src/__main__.py

INFO:     Started server process [3267]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

這次，應該會回傳包含使用者資訊的 200 回應：

# src/models.py
import uuid

import sqlalchemy
from flama.sqlalchemy import metadata
from sqlalchemy.dialects.postgresql import UUID

__all__ = ["user_table", "metadata"]

user_table = sqlalchemy.Table(
    "user",
    metadata,
    sqlalchemy.Column("id", UUID(as_uuid=True), primary_key=True, nullable=False, default=uuid.uuid4),
    sqlalchemy.Column("name", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("surname", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("email", sqlalchemy.String, nullable=False, unique=True),
    sqlalchemy.Column("password", sqlalchemy.String, nullable=False),
    sqlalchemy.Column("active", sqlalchemy.Boolean, nullable=False),
)

使用者停用

最後，我們可以透過使用使用者的憑證向 /user/deactivate/ 發送 POST 請求來停用使用者：

# migrations.py
from sqlalchemy import create_engine

from src.models import metadata

if __name__ == "__main__":
    # Set up the SQLite database
    engine = create_engine("sqlite:///models.db", echo=False)

    # Create the database tables
    metadata.create_all(engine)

    # Print a success message
    print("Database and User table created successfully.")

透過此要求，使用者應該被停用，並且您應該收到帶有空正文的 200 回應。

結論

在這篇文章中，我們深入探討了領域驅動設計 (DDD) 的世界，以及如何在 flama 應用程式中實現它。我們已經了解了 DDD 如何幫助我們將應用程式的業務邏輯與資料存取邏輯分離，以及這種關注點分離如何使程式碼更易於維護和測試。我們還了解如何在flama 應用程式中實現儲存庫和工作模式，以及如何使用它們來封裝資料存取邏輯並提供一種對必須執行的所有操作進行分組的方法在單一事務中的資料來源上。最後，我們了解如何使用資源方法來定義與使用者資料互動的 API 端點，以及如何使用 DDD 模式來實現我們在本範例開頭給出的業務需求。

雖然我們在這裡描述的登入過程並不完全現實，但您可以將本文的資料與先前有關 JWT 身份驗證的文章結合起來，以實現更現實的過程，其中登入最終會返回一個JWT 令牌。如果您對此感興趣，可以使用flama查看JWT身份驗證的帖子。

我們希望您發現這篇文章有用，並且您現在已準備好在自己的 flama 應用程式中實作 DDD。如果您有任何疑問或意見，請隨時與我們聯繫。我們總是很樂意提供協助！

請繼續關注更多關於 flama 以及人工智慧和軟體開發領域其他令人興奮的主題的貼文。下次見！

支持我們的工作

如果您喜歡我們所做的事情，可以透過免費且簡單的方式來支持我們的工作。在 Flama 送給我們 ⭐。

GitHub ⭐ 對我們來說意味著一個世界，它為我們提供了最甜蜜的動力，讓我們繼續努力，幫助其他人踏上構建強大的機器學習 API 的旅程。

您也可以在 ? 上關注我們，我們在這裡分享最新的新聞和更新，以及有關人工智慧、軟體開發等方面的有趣主題。

參考

Flama 文件
Flama GitHub 儲存庫
Flama PyPI 包

關於作者

Vortico：我們專注於軟體開發，幫助企業增強和擴展其人工智慧和技術能力。

以上是使用 Flama 進行本機域驅動設計的詳細內容。更多資訊請關注PHP中文網其他相關文章！

陳述：

本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn

上一篇：如何使用基於跨步的切片從 4D Numpy 數組中有效地裁剪隨機圖像補丁？下一篇：如何使用基於跨步的切片從 4D Numpy 數組中有效地裁剪隨機圖像補丁？

看更多