Natives domänengesteuertes Design mit Flama

Sie haben wahrscheinlich bereits von der jüngsten Version von Flama 1.7 gehört, die einige aufregende neue Funktionen mit sich brachte, die Sie bei der Entwicklung und Produktion Ihrer ML-APIs unterstützen. Dieser Beitrag ist genau einem der wichtigsten Highlights dieser Version gewidmet: Unterstützung für Domain-Driven Design. Bevor wir jedoch anhand eines praktischen Beispiels in die Details eintauchen, empfehlen wir Ihnen, die folgenden Ressourcen im Auge zu behalten (und sich mit ihnen vertraut zu machen, falls Sie dies noch nicht getan haben):

• Offizielle Flama-Dokumentation: Flama-Dokumentation
• Beitrag zur Einführung von Flama für ML-APIs: Einführung in Flama für robuste APIs für maschinelles Lernen

Jetzt beginnen wir mit der neuen Funktion und sehen, wie Sie sie nutzen können, um robuste und wartbare ML-APIs zu erstellen.

Inhaltsverzeichnis

Dieser Beitrag ist wie folgt aufgebaut:

• Was ist Domain-Driven Design?
  • Kurzer Überblick
  • Schlüsselkonzepte
• Implementierung von DDD mit Flama
  • Einrichten der Entwicklungsumgebung
  • Grundanwendung
  • DDD in Aktion
• Fazit
• Unterstützen Sie unsere Arbeit
• Referenzen
• Über die Autoren

Was ist Domain-Driven Design?

Kurzer Überblick

In der modernen Softwareentwicklung ist die Abstimmung der Geschäftslogik mit dem technischen Design einer Anwendung von entscheidender Bedeutung. Hier glänzt Domain-Driven Design (DDD). DDD legt Wert darauf, Software zu entwickeln, die den Kernbereich des Unternehmens widerspiegelt, indem komplexe Probleme durch die Organisation von Code rund um Geschäftskonzepte aufgeschlüsselt werden. Auf diese Weise hilft DDD Entwicklern, wartbare, skalierbare und robuste Anwendungen zu erstellen. Im Folgenden stellen wir die unserer Meinung nach wichtigsten Konzepte von DDD vor, die Sie kennen sollten. Bevor wir näher darauf eingehen, möchten wir anmerken, dass dieser Beitrag weder als umfassender Leitfaden zu DDD noch als Ersatz für die wichtigsten Referenzen zu diesem Thema gedacht ist. Tatsächlich empfehlen wir die folgenden Ressourcen, um ein tieferes Verständnis von DDD zu erlangen:

• Cosmic Python von Harry Percival und Bob Gregory: Dieses Buch ist eine großartige Ressource, um zu lernen, wie man DDD in Python anwendet.
• Domain-Driven Design: Tackling Complexity in the Heart of Software von Eric Evans: Dies ist das Buch, das DDD der Welt vorgestellt hat, und es ist ein Muss für jeden, der ein tiefes Verständnis von DDD entwickeln möchte.

Schlüsselkonzepte

Bevor Sie tiefer in eines der Schlüsselkonzepte von DDD eintauchen, empfehlen wir Ihnen, einen Blick auf eine recht nützliche Abbildung von Cosmic Python zu werfen, in der diese im Kontext einer App dargestellt werden und so zeigen, wie sie miteinander verbunden sind: Abbildung .

Domänenmodell

Das Konzept des Domänenmodells kann durch eine vereinfachte Definition seiner Begriffe erklärt werden:

• Domäne bezieht sich auf den spezifischen Themenbereich der Tätigkeit (oder des Wissens), den unsere Software unterstützen soll.
• Modell bezieht sich auf eine einfache Darstellung (oder Abstraktion) des Systems oder Prozesses, das wir in unserer Software zu kodieren versuchen.

Daher ist das Domänenmodell eine schicke (aber standardmäßige und nützliche) Möglichkeit, sich auf die Reihe von Konzepten und Regeln zu beziehen, die Geschäftsinhaber im Kopf über die Funktionsweise des Unternehmens haben. Dies ist es, was wir auch und allgemein als die Geschäftslogik der Anwendung bezeichnen, einschließlich der Regeln, Einschränkungen und Beziehungen, die das Verhalten des Systems steuern.

Wir werden das Domänenmodell von nun an als Modell bezeichnen.

Repository-Muster

Das Repository-Muster ist ein Entwurfsmuster, das die Entkopplung des Modells vom Datenzugriff ermöglicht. Die Hauptidee des Repository-Musters besteht darin, eine Abstraktionsschicht zwischen der Datenzugriffslogik und der Geschäftslogik einer Anwendung zu erstellen. Diese Abstraktionsschicht ermöglicht die Trennung von Belangen, wodurch der Code wartbarer und testbarer wird.

Bei der Implementierung des Repository-Musters definieren wir normalerweise eine Schnittstelle, die die Standardmethoden angibt, die jedes andere Repository implementieren muss (AbstractRepository). Anschließend wird mit der konkreten Implementierung dieser Methoden ein bestimmtes Repository definiert, in dem die Datenzugriffslogik implementiert ist (z. B. SQLAlchemyRepository). Dieses Entwurfsmuster zielt darauf ab, die Datenmanipulationsmethoden zu isolieren, sodass sie nahtlos an anderer Stelle in der Anwendung verwendet werden können, z. B. in unserem Domänenmodell.

Arbeitseinheitsmuster

Das Unit-of-Work-Muster ist der fehlende Teil, um das Modell endgültig vom Datenzugriff zu entkoppeln. Die Arbeitseinheit kapselt die Datenzugriffslogik und bietet eine Möglichkeit, alle Vorgänge zu gruppieren, die an der Datenquelle in einer einzigen Transaktion ausgeführt werden müssen. Dieses Muster stellt sicher, dass alle Operationen atomar ausgeführt werden.

Bei der Implementierung des Arbeitseinheitsmusters definieren wir normalerweise eine Schnittstelle, die die Standardmethoden angibt, die jede andere Arbeitseinheit implementieren muss (AbstractUnitOfWork). Und dann wird eine bestimmte Arbeitseinheit mit der konkreten Implementierung dieser Methoden definiert, in der die Datenzugriffslogik implementiert ist (z. B. SQLAlchemyUnitOfWork). Dieses Design ermöglicht eine systematische Handhabung der Verbindung zur Datenquelle, ohne dass die Implementierung der Geschäftslogik der Anwendung geändert werden muss.

Implementierung von DDD mit Flama

Nach der kurzen Einführung in die Hauptkonzepte von DDD sind wir bereit, mit Flama in die Implementierung von DDD einzutauchen. In diesem Abschnitt führen wir Sie durch den Prozess der Einrichtung der Entwicklungsumgebung, der Erstellung einer Basisanwendung und der Implementierung von DDD-Konzepten mit Flama.

Bevor Sie mit dem Beispiel fortfahren, werfen Sie bitte einen Blick auf die Namenskonvention von Flama bezüglich der wichtigsten DDD-Konzepte, die wir gerade überprüft haben:

Wie Sie in der Abbildung oben sehen können, ist die Namenskonvention recht intuitiv: Repository bezieht sich auf das Repository-Muster; und „Arbeiter“ bezieht sich auf die Arbeitseinheit. Jetzt können wir mit der Implementierung einer Flama-API fortfahren, die DDD verwendet. Bevor wir jedoch beginnen, sollten Sie die Grundlagen zum Erstellen einer einfachen API mit flama oder zum Ausführen der API überprüfen, sobald der Code bereits fertig ist. Vielleicht möchten Sie dies überprüfen Lesen Sie die Kurzanleitung durch. Dort finden Sie die grundlegenden Konzepte und Schritte, die zum Befolgen dieses Beitrags erforderlich sind. Beginnen wir nun ohne weitere Umschweife mit der Umsetzung.

Einrichten der Entwicklungsumgebung

Unser erster Schritt besteht darin, unsere Entwicklungsumgebung zu erstellen und alle erforderlichen Abhängigkeiten für dieses Projekt zu installieren. Das Gute daran ist, dass wir für dieses Beispiel nur flama installieren müssen, um über alle notwendigen Tools zur Implementierung der JWT-Authentifizierung zu verfügen. Wir werden Poesie verwenden, um unsere Abhängigkeiten zu verwalten, aber Sie können auch pip verwenden, wenn Sie möchten:

poetry add "flama[full]" "aiosqlite"

Das aiosqlite-Paket ist erforderlich, um SQLite mit SQLAlchemy zu verwenden, der Datenbank, die wir in diesem Beispiel verwenden werden.

Wenn Sie wissen möchten, wie wir unsere Projekte normalerweise organisieren, schauen Sie sich unseren vorherigen Beitrag hier an, in dem wir ausführlich erklären, wie man ein Python-Projekt mit Poesie einrichtet und welche Projektordnerstruktur wir normalerweise befolgen.

Basisanwendung

Beginnen wir mit einer einfachen Anwendung, die über einen einzigen öffentlichen Endpunkt verfügt. Dieser Endpunkt gibt eine kurze Beschreibung der API zurück.

# 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}

Wenn Sie diese Anwendung ausführen möchten, können Sie den obigen Code in einer Datei namens app.py im Ordner src speichern und dann den folgenden Befehl ausführen (denken Sie daran, die Poesie-Umgebung aktiviert zu haben, andernfalls müssen Sie dies tun Stellen Sie dem Befehl „poetry run“) voran):

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)

wobei das Flag --server-reload optional ist und verwendet wird, um den Server automatisch neu zu laden, wenn sich der Code ändert. Dies ist während der Entwicklung sehr nützlich, Sie können es jedoch entfernen, wenn Sie es nicht benötigen. Für eine vollständige Liste der verfügbaren Optionen können Sie flama run --help ausführen oder die Dokumentation überprüfen.

Alternativ können Sie die Anwendung auch ausführen, indem Sie das folgende Skript ausführen, das Sie als __main__.py im Ordner src speichern können:

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

Und dann können Sie die Anwendung ausführen, indem Sie den folgenden Befehl ausführen:

poetry add "flama[full]" "aiosqlite"

DDD in Aktion

Nachdem wir nun ein minimales Grundgerüst für unsere Anwendung eingerichtet haben, können wir mit der Implementierung der DDD-Konzepte beginnen, die wir gerade im
überprüft haben Kontext eines einfachen Beispiels, das versucht, ein reales Szenario nachzuahmen. Nehmen wir an, wir werden gebeten, eine API zur Verwaltung von Benutzern zu entwickeln, und wir erhalten die folgenden Anforderungen:

• Wir möchten neue Benutzer über eine POST-Anfrage an /user/ erstellen und dabei den Vor- und Nachnamen, die E-Mail-Adresse und das Passwort des Benutzers angeben.
• Jeder erstellte Benutzer wird in einer Datenbank mit dem folgenden Schema gespeichert:
  • id: eindeutige Kennung für den Benutzer.
  • Name: Benutzername.
  • Nachname: Nachname des Benutzers.
  • E-Mail: E-Mail des Benutzers.
  • Passwort: Passwort des Benutzers. Dies sollte gehasht werden, bevor es in der Datenbank gespeichert wird.
  • aktiv: ein boolesches Flag, das angibt, ob der Benutzer aktiv ist oder nicht. Standardmäßig werden Benutzer als inaktiv erstellt.
• Erstellte Benutzer müssen ihr Konto aktivieren, indem sie mit ihrer E-Mail-Adresse und ihrem Passwort eine POST-Anfrage an /user/activate/ senden. Sobald der Benutzer aktiviert ist, muss der Status des Benutzers in der Datenbank auf „Aktiv“ aktualisiert werden.
• Benutzer können sich anmelden, indem sie mit ihrer E-Mail-Adresse und ihrem Passwort eine POST-Anfrage an /user/signin/ senden. Wenn der Benutzer aktiv ist, muss die API alle Benutzerinformationen zurückgeben. Andernfalls muss die API eine Fehlermeldung zurückgeben.
• Benutzer, die ihr Konto deaktivieren möchten, können dies tun, indem sie mit ihrer E-Mail-Adresse und ihrem Passwort eine POST-Anfrage an /user/deactivate/ senden. Sobald der Benutzer deaktiviert ist, muss der Status des Benutzers in der Datenbank auf inaktiv aktualisiert werden.

Dieser Satz von Anforderungen stellt das dar, was wir zuvor als Domänenmodell unserer Anwendung bezeichnet haben, das im Wesentlichen nichts anderes als eine Materialisierung des folgenden Benutzerworkflows ist:

1. Ein Benutzer wird über eine POST-Anfrage an /user/ erstellt.
2. Der Benutzer aktiviert sein Konto über eine POST-Anfrage an /user/activate/.
3. Der Benutzer meldet sich über eine POST-Anfrage bei /user/signin/ an.
4. Der Benutzer deaktiviert sein Konto über eine POST-Anfrage an /user/deactivate/.
5. Der Benutzer kann die Schritte 2–4 so oft wiederholen, wie er möchte.

Jetzt implementieren wir das Domänenmodell mithilfe der Repository- und Worker-Muster. Wir beginnen mit der Definition des Datenmodells und implementieren dann die Repository- und Worker-Muster.

Datenmodell

Unsere Benutzerdaten werden in einer SQLite-Datenbank gespeichert (Sie können jede andere von SQLAlchemy unterstützte Datenbank verwenden). Wir verwenden das folgende Datenmodell zur Darstellung der Benutzer (Sie können diesen Code in einer Datei namens models.py im Ordner src speichern):

# 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}

Neben dem Datenmodell benötigen wir ein Migrationsskript, um die Datenbank und die Tabelle zu erstellen. Dazu können wir den folgenden Code in einer Datei namens migrations.py im Stammverzeichnis des Projekts speichern:

poetry add "flama[full]" "aiosqlite"

Und dann können wir das Migrationsskript ausführen, indem wir den folgenden Befehl ausführen:

# 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}

Repository

In diesem Beispiel benötigen wir nur ein Repository, nämlich das Repository, das die atomaren Operationen in der Benutzertabelle abwickelt, dessen Name UserRepository sein wird. Glücklicherweise bietet flama eine Basisklasse für Repositorys im Zusammenhang mit SQLAlchemy-Tabellen, genannt SQLAlchemyTableRepository.

Die Klasse SQLAlchemyTableRepository stellt eine Reihe von Methoden zur Durchführung von CRUD-Operationen für die Tabelle bereit, insbesondere:

• create: Erstellt neue Elemente in der Tabelle. Wenn das Element bereits vorhanden ist, wird eine Ausnahme (IntegrityError) ausgelöst, andernfalls wird der Primärschlüssel des neuen Elements zurückgegeben.
• Abrufen: Ruft ein Element aus der Tabelle ab. Wenn das Element nicht existiert, wird eine Ausnahme (NotFoundError) ausgelöst, andernfalls wird das Element zurückgegeben. Wenn mehr als ein Element gefunden wird, wird eine Ausnahme ausgelöst (MultipleRecordsError).
• update: Aktualisiert ein Element in der Tabelle. Wenn das Element nicht vorhanden ist, wird eine Ausnahme (NotFoundError) ausgelöst, andernfalls wird das aktualisierte Element zurückgegeben.
• delete: Löscht ein Element aus der Tabelle.
• Liste: Listet alle Elemente in der Tabelle auf, die den übergebenen Klauseln und Filtern entsprechen. Wenn keine Klauseln oder Filter angegeben sind, werden alle Elemente in der Tabelle zurückgegeben. Wenn keine Elemente gefunden werden, wird eine leere Liste zurückgegeben.
• drop: Löscht die Tabelle aus der Datenbank.

Für die Zwecke unseres Beispiels benötigen wir keine weitere Aktion an der Tabelle, daher sind die vom SQLAlchemyTableRepository bereitgestellten Methoden ausreichend. Wir können den folgenden Code in einer Datei namens repositories.py im Ordner src speichern:

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)

Wie Sie sehen können, ist die UserRepository-Klasse eine Unterklasse von SQLAlchemyTableRepository und erfordert nur, dass die Tabelle im _table-Attribut festgelegt wird. Dies ist das Einzige, was wir tun müssen, um ein voll funktionsfähiges Repository für die Benutzertabelle zu haben.

Wenn wir über die Standard-CRUD-Operationen hinaus benutzerdefinierte Methoden hinzufügen möchten, können wir dies tun, indem wir sie in der UserRepository-Klasse definieren. Wenn wir beispielsweise eine Methode zum Zählen der Anzahl aktiver Benutzer hinzufügen möchten, könnten wir dies wie folgt tun:

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

Obwohl wir diese Methode in unserem Beispiel nicht verwenden werden, ist es gut zu wissen, dass wir bei Bedarf benutzerdefinierte Methoden zum Repository hinzufügen können und wie diese implementiert werden
im Kontext des Repository-Musters. Wie wir bereits sehen können, ist dies ein leistungsstarkes Entwurfsmuster, da wir hier die gesamte Datenzugriffslogik implementieren können, ohne die Geschäftslogik der Anwendung ändern zu müssen (die in den entsprechenden Ressourcenmethoden implementiert ist).

Arbeitnehmer

Das Unit-of-Work-Muster wird verwendet, um die Datenzugriffslogik zu kapseln und eine Möglichkeit zu bieten, alle Vorgänge, die an der Datenquelle ausgeführt werden müssen, in einer einzigen Transaktion zu gruppieren. In flama ist das UoW-Muster mit dem Namen Worker implementiert. Auf die gleiche Weise wie beim Repository-Muster stellt flama eine Basisklasse für Worker bereit, die sich auf SQLAlchemy-Tabellen beziehen, namens SQLAlchemyWorker. Im Wesentlichen stellt der SQLAlchemyWorker eine Verbindung und eine Transaktion zur Datenbank bereit und instanziiert alle seine Repositorys mit der Worker-Verbindung. In diesem Beispiel verwendet unser Worker nur ein einziges Repository (nämlich das UserRepository), aber wir könnten bei Bedarf weitere Repositorys hinzufügen.

Unser Worker heißt RegisterWorker und wir können den folgenden Code in einer Datei namens Workers.py im Ordner src speichern:

poetry add "flama[full]" "aiosqlite"

Wenn wir also mehr Repositorys zum Arbeiten hätten, zum Beispiel ProductRepository und OrderRepository, könnten wir diese wie folgt zum Worker hinzufügen:

# 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}

So einfach ist das: Wir haben die Repository- und Worker-Muster in unserer Anwendung implementiert. Jetzt können wir mit der Implementierung der Ressourcenmethoden fortfahren, die die API-Endpunkte bereitstellen, die für die Interaktion mit den Benutzerdaten erforderlich sind.

Ressourcen

Ressourcen sind einer der Hauptbausteine ​​einer flama-Anwendung. Sie dienen zur Darstellung von Anwendungsressourcen (im Sinne von RESTful-Ressourcen) und zur Definition der API-Endpunkte, die mit ihnen interagieren.

In unserem Beispiel definieren wir eine Ressource für den Benutzer namens UserResource, die die Methoden zum Erstellen, Aktivieren, Anmelden und Deaktivieren von Benutzern enthält. Ressourcen müssen zumindest von der integrierten Ressourcenklasse flama abgeleitet werden, obwohl flama anspruchsvollere Klassen für die Arbeit bereitstellt, z. B. RESTResource und CRUDResource.

Wir können den folgenden Code in einer Datei namens resources.py im Ordner src speichern:

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)

Basisanwendung mit DDD

Nachdem wir nun das Datenmodell, die Repository- und Worker-Muster sowie die Ressourcenmethoden implementiert haben, müssen wir die zuvor eingeführte Basisanwendung ändern, damit alles wie erwartet funktioniert. Wir müssen:

• Fügen Sie die SQLAlchemy-Verbindung zur Anwendung hinzu. Dies wird erreicht, indem Sie das SQLAlchemyModule als Modul zum Anwendungskonstruktor hinzufügen.
• Fügen Sie den Worker zur Anwendung hinzu. Dies wird erreicht, indem Sie den RegisterWorker als Komponente zum Anwendungskonstruktor hinzufügen.

Dadurch bleibt die app.py-Datei wie folgt:

poetry add "flama[full]" "aiosqlite"

Es sollte Ihnen bereits klar sein, wie das DDD-Muster es uns ermöglicht hat, die Geschäftslogik der Anwendung (die in den Ressourcenmethoden leicht lesbar ist) von der Datenzugriffslogik (was in den Repository- und Worker-Mustern implementiert ist). Es ist auch erwähnenswert, wie diese Trennung von Belangen den Code wartbarer und testbarer gemacht hat und wie der Code jetzt besser auf die Geschäftsanforderungen abgestimmt ist, die uns zu Beginn dieses Beispiels gegeben wurden.

Ausführen der Anwendung

Bevor Sie einen Befehl ausführen, überprüfen Sie bitte, ob Ihre Entwicklungsumgebung korrekt eingerichtet ist und die Ordnerstruktur wie folgt lautet:

# 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}

Wenn alles richtig eingerichtet ist, können Sie die Anwendung ausführen, indem Sie den folgenden Befehl ausführen (denken Sie daran, das Migrationsskript auszuführen, bevor Sie die Anwendung ausführen):

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)

Jetzt können wir die Geschäftslogik ausprobieren, die wir gerade implementiert haben. Denken Sie daran, dass Sie dies entweder mit einem Tool wie Curl oder Postman versuchen können oder indem Sie die von

flama automatisch generierte Benutzeroberfläche für Dokumente verwenden, indem Sie in Ihrem Browser zu http://localhost:8000/docs/ navigieren und von dort aus die Endpunkte ausprobieren.

Erstellen Sie einen Benutzer

Um einen Benutzer zu erstellen, können Sie eine POST-Anfrage an /user/ mit der folgenden Nutzlast senden:

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

Wir können also Curl verwenden, um die Anfrage wie folgt zu senden:

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)

Wenn die Anfrage erfolgreich ist, sollten Sie eine 200-Antwort mit einem leeren Text erhalten und der Benutzer wird in der Datenbank erstellt.

anmelden

Um sich anzumelden, können Sie eine POST-Anfrage an /user/signin/ mit der folgenden Nutzlast senden:

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

Wir können also Curl verwenden, um die Anfrage wie folgt zu senden:

# 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.")

Da der Benutzer nicht aktiv ist, sollten Sie etwa die folgende Antwort erhalten:

> poetry run python migrations.py

Database and User table created successfully.

Wir können auch testen, was passieren würde, wenn jemand versucht, sich mit dem falschen Passwort anzumelden:

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

from src import models

__all__ = ["UserRepository"]

class UserRepository(SQLAlchemyTableRepository):
    _table = models.user_table

In diesem Fall sollten Sie eine 401-Antwort mit folgendem Text erhalten:

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

Abschließend sollten wir auch versuchen, uns mit einem Benutzer anzumelden, der nicht existiert:

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

from src import repositories

__all__ = ["RegisterWorker"]


class RegisterWorker(SQLAlchemyWorker):
    user: repositories.UserRepository

In diesem Fall sollten Sie eine 404-Antwort mit folgendem Text erhalten:

# 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

Benutzeraktivierung

Nachdem wir den Anmeldevorgang untersucht haben, können wir nun den Benutzer aktivieren, indem wir eine POST-Anfrage an /user/activate/ mit den Anmeldeinformationen des Benutzers senden:

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

Mit dieser Anfrage sollte der Benutzer aktiviert werden und Sie sollten eine 200-Antwort mit einem leeren Text erhalten.

Wie im vorherigen Fall können wir auch testen, was passieren würde, wenn jemand versucht, den Benutzer mit dem falschen Passwort zu aktivieren:

poetry add "flama[full]" "aiosqlite"

In diesem Fall sollten Sie eine 401-Antwort mit folgendem Text erhalten:

# 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}

Abschließend sollten wir auch versuchen, einen Benutzer zu aktivieren, der nicht existiert:

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)

In diesem Fall sollten Sie eine 404-Antwort mit folgendem Text erhalten:

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

Benutzeranmeldung nach der Aktivierung

Da der Benutzer nun aktiviert ist, können wir versuchen, uns erneut anzumelden:

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)

Diesmal sollte eine 200-Antwort mit den Benutzerinformationen zurückgegeben werden:

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

Benutzerdeaktivierung

Schließlich können wir den Benutzer deaktivieren, indem wir eine POST-Anfrage an /user/deactivate/ mit den Anmeldeinformationen des Benutzers senden:

# 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.")

Mit dieser Anfrage sollte der Benutzer deaktiviert werden und Sie sollten eine 200-Antwort mit einem leeren Text erhalten.

Abschluss

In diesem Beitrag haben wir uns in die Welt des Domain-Driven Design (DDD) gewagt und wie es in einer flama-Anwendung implementiert werden kann. Wir haben gesehen, wie DDD uns helfen kann, die Geschäftslogik der Anwendung von der Datenzugriffslogik zu trennen, und wie diese Trennung von Belangen den Code wartbarer und testbarer machen kann. Wir haben auch gesehen, wie die Repository- und Worker-Muster in einer flama-Anwendung implementiert werden können und wie sie verwendet werden können, um die Datenzugriffslogik zu kapseln und eine Möglichkeit zu bieten, alle auszuführenden Vorgänge zu gruppieren auf der Datenquelle innerhalb einer einzigen Transaktion. Schließlich haben wir gesehen, wie die Ressourcenmethoden verwendet werden können, um die API-Endpunkte zu definieren, die mit den Benutzerdaten interagieren, und wie das DDD-Muster verwendet werden kann, um die Geschäftsanforderungen zu implementieren, die uns zu Beginn dieses Beispiels gegeben wurden.

Obwohl der hier beschriebene Anmeldevorgang nicht ganz realistisch ist, könnten Sie das Material dieses und eines früheren Beitrags zur JWT-Authentifizierung kombinieren, um einen realistischeren Prozess zu implementieren, bei dem die Anmeldung am Ende eine zurückgibt JWT-Token. Wenn Sie daran interessiert sind, können Sie sich den Beitrag zur JWT-Authentifizierung mit flama ansehen.

Wir hoffen, dass Sie diesen Beitrag nützlich fanden und dass Sie nun bereit sind, DDD in Ihren eigenen flama-Anwendungen zu implementieren. Wenn Sie Fragen oder Anmerkungen haben, können Sie sich gerne an uns wenden. Wir helfen Ihnen gerne weiter!

Seien Sie gespannt auf weitere Beiträge zu flama und anderen spannenden Themen in der Welt der KI und Softwareentwicklung. Bis zum nächsten Mal!

Unterstützen Sie unsere Arbeit

Wenn Ihnen gefällt, was wir tun, gibt es eine kostenlose und einfache Möglichkeit, unsere Arbeit zu unterstützen. Schenkt uns ein ⭐ bei Flama.

GitHub ⭐ bedeutet uns eine Welt und gibt uns den besten Treibstoff, weiter daran zu arbeiten und anderen auf ihrem Weg zur Entwicklung robuster APIs für maschinelles Lernen zu helfen.

Sie können uns auch auf ? folgen, wo wir neben interessanten Threads zu KI, Softwareentwicklung und vielem mehr unsere neuesten Nachrichten und Updates teilen.

Referenzen

• Flama-Dokumentation
• Flama GitHub-Repository
• Flama PyPI-Paket

Über die Autoren

• Vortico: Wir sind auf Softwareentwicklung spezialisiert, um Unternehmen bei der Verbesserung und Erweiterung ihrer KI- und Technologiefähigkeiten zu unterstützen.

Das obige ist der detaillierte Inhalt vonNatives domänengesteuertes Design mit Flama. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!