ホームページ > 記事 > バックエンド開発 > Flama を使用したネイティブドメイン駆動設計

Flama を使用したネイティブドメイン駆動設計

Barbara Streisandオリジナル: 2024-11-03 14:08:03624ブラウズ

おそらく、Flama 1.7 の最近のリリースについてはすでに聞いたことがあるでしょう。これは、ML API の開発と製品化に役立ついくつかのエキサイティングな新機能をもたらしました。この投稿は、そのリリースの主なハイライトの 1 つである ドメイン駆動設計のサポート に特化しています。ただし、実際の例で詳細を説明する前に、次のリソースに留意することをお勧めします (まだ理解していない場合は、よく理解しておくこと)。

Flama の公式ドキュメント: Flama ドキュメント
ML API 向け Flama の紹介の投稿: 堅牢な機械学習 API のための Flama の紹介

それでは、新機能の使用を開始し、それを活用して堅牢で保守可能な ML API を構築する方法を見てみましょう。

ドメイン駆動設計とは何ですか?

簡単な概要

現代のソフトウェア開発では、ビジネスロジックをアプリケーションの技術設計と調整することが不可欠です。ここでドメイン駆動設計 (DDD) が威力を発揮します。 DDD は、ビジネスの中核領域を反映するソフトウェアの構築に重点を置き、ビジネス概念に基づいてコードを整理することで複雑な問題を解決します。これにより、DDD は開発者が保守可能でスケーラブルで堅牢なアプリケーションを作成できるようにします。以下では、知っておくべき DDD の最も重要な概念を私たちが考えるものを紹介します。それらについて詳しく説明する前に、この投稿は DDD の包括的なガイドを目的としたものではなく、またこのトピックに関する主要な参考文献の代替を目的としたものではないことを述べておきます。実際、DDD をより深く理解するには、次のリソースをお勧めします:

『Cosmic Python』Harry Percival と Bob Gregory 著: この本は、Python で DDD を適用する方法を学ぶための素晴らしいリソースです。
『ドメイン駆動設計: ソフトウェアの中心部の複雑さへの取り組み』エリック・エヴァンス著: これは DDD を世界に紹介した本であり、DDD を深く理解したい人にとって必読の書です。

主要な概念

DDD の重要な概念をさらに詳しく説明する前に、これらがアプリのコンテキストで示され、どのように相互接続されているかを示す、Cosmic Python による非常に役立つ図を参照することをお勧めします。 .

ドメインモデル

ドメインモデル の概念は、その用語の単純な定義によって説明できます。

ドメイン は、ソフトウェアがサポートするために構築されているアクティビティ (または知識) の特定の主題領域を指します。
モデル は、ソフトウェアでエンコードしようとしているシステムまたはプロセスの単純な表現 (または抽象化) を指します。

したがって、ドメインモデルは、ビジネスの仕組みについてビジネスオーナーが頭の中に持つ一連の概念とルールを参照するための派手な (しかし標準的で便利な) 方法です。これは、一般に、アプリケーションのビジネスロジックとも呼ばれるもので、システムの動作を制御するルール、制約、関係が含まれます。

今後、ドメインモデルをモデルと呼びます。

リポジトリパターン

リポジトリパターンは、モデルをデータアクセスから切り離すことを可能にする設計パターンです。リポジトリパターンの背後にある主なアイデアは、データアクセスロジックとアプリケーションのビジネスロジックの間に抽象化レイヤーを作成することです。この抽象化レイヤーにより関心事の分離が可能になり、コードの保守性とテスト性が向上します。

リポジトリパターンを実装するときは、通常、他のリポジトリが実装する必要がある標準メソッド (AbstractRepository) を指定するインターフェイスを定義します。そして、データアクセスロジックが実装されるこれらのメソッドの具体的な実装を使用して、特定のリポジトリが定義されます (SQLAlchemyRepository など)。この設計パターンは、データ操作メソッドを分離して、アプリケーション内の他の場所でシームレスに使用できるようにすることを目的としています。私たちのドメインモデルで。

作業単位パターン

作業単位パターン は、モデルをデータアクセスから最終的に切り離すために欠落している部分です。作業単位はデータアクセスロジックをカプセル化し、単一のトランザクション内でデータソースに対して実行する必要があるすべての操作をグループ化する方法を提供します。このパターンでは、すべての操作がアトミックに実行されることが保証されます。

作業単位パターンを実装するときは、通常、他の作業単位が実装する必要がある標準メソッド (AbstractUnitOfWork) を指定するインターフェイスを定義します。そして、データアクセスロジックが実装されるこれらのメソッドの具体的な実装を使用して、特定の作業単位が定義されます (SQLAlchemyUnitOfWork など)。この設計により、アプリケーションのビジネスロジックの実装を変更することなく、データソースへの接続を体系的に処理できるようになります。

Flama を使用した DDD の実装

DDD の主な概念を簡単に紹介した後、Flama を使用した DDD の実装に入る準備が整いました。このセクションでは、開発環境のセットアップ、ベースアプリケーションの構築、Flama を使用した DDD コンセプトの実装のプロセスを説明します。

例に進む前に、今確認した主要な DDD 概念に関する Flama の命名規則を確認してください。

Native Domain-Driven Design with Flama

上の図からわかるように、命名規則は非常に直感的です。リポジトリはリポジトリパターンを指します。そして、Worker は作業単位を指します。ここで、DDD を使用する Flama API の実装に進むことができます。ただし、始める前に、flama を使用して単純な API を作成する方法、またはコードの準備ができた後に API を実行する方法についての基本を確認する必要がある場合は、次のことを確認してください。クイックスタートガイドを出してください。ここには、この投稿を進めるために必要な基本的な概念と手順が記載されています。それでは、早速、実装を始めましょう。

開発環境のセットアップ

最初のステップは、開発環境を作成し、このプロジェクトに必要な依存関係をすべてインストールすることです。良い点は、この例では、flama をインストールするだけで、JWT 認証を実装するために必要なツールがすべて揃っていることです。依存関係を管理するために詩を使用しますが、必要に応じて pip を使用することもできます:

poetry add "flama[full]" "aiosqlite"

aiosqlite パッケージは、この例で使用するデータベースである SQLAlchemy で SQLite を使用するために必要です。

私たちが通常どのようにプロジェクトを編成しているかを知りたい場合は、こちらの前回の投稿をご覧ください。そこでは、詩を使用した 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 というファイルに保存し、次のコマンドを実行します (詩環境をアクティブ化することを忘れないでください。アクティブ化していない場合は、コマンドの前に詩 run を付けます):

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 リクエストを送信して、アカウントをアクティブ化する必要があります。ユーザーがアクティブ化されたら、データベース内のユーザーのステータスをアクティブに更新する必要があります。
ユーザーは、電子メールとパスワードを使用して POST リクエストを /user/signin/ に送信することでサインインできます。ユーザーがアクティブな場合、API はすべてのユーザー情報を返す必要があります。それ以外の場合、API はエラーメッセージを返す必要があります。
アカウントを無効化したいユーザーは、電子メールとパスワードを使用して /user/deactivate/ に POST リクエストを送信することで無効化できます。ユーザーが非アクティブ化されたら、データベース内のユーザーのステータスを非アクティブに更新する必要があります。

この一連の要件は、これまでアプリケーションの ドメインモデル と呼ばれていたものを構成します。これは本質的には、次のユーザーワークフローを具体化したものに他なりません。

ユーザーは /user/ への POST リクエストによって作成されます。
ユーザーは、/user/activate/ への POST リクエストを介してアカウントをアクティブ化します。
ユーザーは、POST リクエストを介して /user/signin/ にサインインします。
ユーザーは、/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}

リポジトリ

この例では、必要なリポジトリは 1 つだけです。つまり、ユーザーテーブルでのアトミック操作を処理するリポジトリであり、名前は UserRepository になります。ありがたいことに、flama は SQLAlchemyTableRepository という SQLAlchemy テーブルに関連するリポジトリの基本クラスを提供します。

クラス SQLAlchemyTableRepository は、テーブルに対して CRUD 操作を実行する一連のメソッドを提供します。具体的には次のとおりです。

create: テーブルに新しい要素を作成します。要素が既に存在する場合は例外 (IntegrityError) が発生し、それ以外の場合は新しい要素の主キーが返されます。
retrieve: テーブルから要素を取得します。要素が存在しない場合は例外 (NotFoundError) が発生し、存在しない場合は要素が返されます。複数の要素が見つかった場合は、例外 (MultipleRecordsError) が発生します。
update: テーブル内の要素を更新します。要素が存在しない場合は例外 (NotFoundError) が発生し、存在しない場合は更新された要素が返されます。
delete: テーブルから要素を削除します。
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 は SQLAlchemyWorker と呼ばれる、SQLAlchemy テーブルに関連するワーカーの基本クラスを提供します。基本的に、SQLAlchemyWorker はデータベースへの接続とトランザクションを提供し、ワーカー接続を使用してすべてのリポジトリをインスタンス化します。この例では、ワーカーは 1 つのリポジトリ (つまり UserRepository) のみを使用しますが、必要に応じてさらにリポジトリを追加することもできます。

ワーカーは RegisterWorker という名前になり、次のコードを src フォルダーの下の works.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 アプリケーションの主要な構成要素の 1 つです。これらは、アプリケーションリソース (RESTful リソースの意味で) を表し、それらと対話する API エンドポイントを定義するために使用されます。

この例では、UserResource というユーザーのリソースを定義します。これには、ユーザーを作成、アクティブ化、サインイン、および非アクティブ化するためのメソッドが含まれます。リソースは、少なくとも flama 組み込み Resource クラスから派生する必要がありますが、flama は、RESTResource や CRUDResource など、操作するためのより高度なクラスを提供します。

次のコードを src フォルダーの下の resource.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 をアプリケーションコンストラクターにモジュールとして追加することで実現されます。
ワーカーをアプリケーションに追加します。これは、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

ユーザーを作成する

ユーザーを作成するには、次のペイロードを含む POST リクエストを /user/ に送信できます。

# 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 レスポンスが返され、データベースにユーザーが作成されます。

サインイン

サインインするには、次のペイロードを使用して POST リクエストを /user/signin/ に送信できます。

# 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

ユーザーのアクティベーション

サインインプロセスを調べたので、ユーザーの資格情報を使用して POST リクエストを /user/activate/ に送信することでユーザーをアクティブ化できます。

# 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),
)

ユーザーの無効化

最後に、ユーザーの資格情報を使用して POST リクエストを /user/deactivate/ に送信することで、ユーザーを非アクティブ化できます。

# 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 を実装する準備ができたことを願っています。ご質問やご意見がございましたら、お気軽にお問い合わせください。いつでも喜んでお手伝いさせていただきます!

フラマや、AI とソフトウェア開発の世界におけるその他のエキサイティングなトピックに関するさらなる投稿にご期待ください。次回まで！

私たちの仕事をサポートしてください

私たちの活動を気に入っていただけましたら、無料で簡単に私たちの活動をサポートする方法があります。 Flama で ⭐ をプレゼントしてください。

GitHub ⭐ は私たちにとって世界を意味し、堅牢な機械学習 API を構築する旅の他の人々を支援するために、GitHub に取り組み続けるための最もおいしい燃料を与えてくれます。

? で私たちをフォローすることもできます。そこでは、AI、ソフトウェア開発などに関する興味深いスレッドのほかに、最新のニュースや更新情報を共有しています。

参考文献

Flame ドキュメント
Flama GitHub リポジトリ
Flama PyPI パッケージ

著者について

Vortico: 私たちは、企業が AI とテクノロジーの能力を強化および拡張できるよう支援するソフトウェア開発を専門としています。

以上がFlama を使用したネイティブドメイン駆動設計の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

声明：

この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。

前の記事：ストライドベースのスライスを使用して 4D Numpy 配列からランダムな画像パッチを効率的に切り出すにはどうすればよいですか?次の記事：ストライドベースのスライスを使用して 4D Numpy 配列からランダムな画像パッチを効率的に切り出すにはどうすればよいですか?

続きを見る

Flama を使用したネイティブ ドメイン駆動設計

目次