Conception native basée sur le domaine avec Flama

Vous avez probablement déjà entendu parler de la récente version de Flama 1.7, qui a apporté de nouvelles fonctionnalités intéressantes pour vous aider dans le développement et la production de vos API ML. Cet article est précisément consacré à l'un des principaux points forts de cette version : Support for Domain-Driven Design. Mais, avant d'entrer dans les détails avec un exemple pratique, nous vous recommandons de garder à l'esprit les ressources suivantes (et de vous familiariser avec elles si ce n'est pas déjà fait) :

• Documentation officielle Flama : Documentation Flama
• Post présentant Flama pour les API ML : Introduction à Flama pour des API robustes de Machine Learning

Maintenant, commençons avec la nouvelle fonctionnalité et voyons comment vous pouvez l'exploiter pour créer des API ML robustes et maintenables.

Table des matières

Ce message est structuré comme suit :

• Qu’est-ce que la conception pilotée par domaine ?
  • Bref aperçu
  • Concepts clés
• Implémentation de DDD avec Flama
  • Mise en place de l'environnement de développement
  • Application de base
  • DDD en action
• Conclusion
• Soutenez notre travail
• Références
• À propos des auteurs

Qu’est-ce que la conception pilotée par domaine ?

Bref aperçu

Dans le développement de logiciels modernes, il est essentiel d'aligner la logique métier avec la conception technique d'une application. C’est là que la conception pilotée par domaine (DDD) brille. DDD met l'accent sur la création de logiciels qui reflètent le domaine principal de l'entreprise, en décomposant les problèmes complexes en organisant le code autour de concepts commerciaux. Ce faisant, DDD aide les développeurs à créer des applications maintenables, évolutives et robustes. Dans ce qui suit, nous présentons ce que nous considérons comme les concepts les plus importants du DDD que vous devez connaître. Avant de les aborder, remarquons que cet article n'est pas destiné à être un guide complet de DDD, ni à se substituer aux principales références sur le sujet. En effet, nous recommandons les ressources suivantes pour mieux comprendre DDD :

• Cosmic Python par Harry Percival et Bob Gregory : ce livre est une excellente ressource pour apprendre à appliquer DDD en Python.
• Domain-Driven Design : Tackling Complexity in the Heart of Software par Eric Evans : c'est le livre qui a présenté DDD au monde, et c'est une lecture incontournable pour quiconque souhaite développer une compréhension approfondie de DDD.

Concepts clés

Avant de plonger plus profondément dans l'un des concepts clés de DDD, nous vous recommandons de jeter un œil à une figure très utile de Cosmic Python où ceux-ci sont affichés dans le contexte d'une application, montrant ainsi comment ils sont interconnectés : figure .

Modèle de domaine

Le concept de modèle de domaine peut s'expliquer par une définition simpliste de ses termes :

• domaine fait référence au domaine d'activité (ou de connaissances) spécifique pour lequel notre logiciel est conçu pour prendre en charge.
• modèle fait référence à une simple représentation (ou abstraction) du système ou du processus que nous essayons d'encoder dans notre logiciel.

Ainsi, le modèle de domaine est une manière sophistiquée (mais standard et utile) de faire référence à l'ensemble de concepts et de règles que les propriétaires d'entreprise ont en tête sur le fonctionnement de l'entreprise. C'est ce que nous appelons aussi et communément la logique métier de l'application, y compris les règles, les contraintes et les relations qui régissent le comportement du système.

Nous appellerons désormais le modèle de domaine le modèle.

Modèle de référentiel

Le modèle de référentiel est un modèle de conception qui permet de découpler le modèle de l'accès aux données. L'idée principale derrière le modèle de référentiel est de créer une couche d'abstraction entre la logique d'accès aux données et la logique métier d'une application. Cette couche d'abstraction permet de séparer les préoccupations, rendant le code plus maintenable et testable.

Lors de l'implémentation du modèle de référentiel, nous définissons généralement une interface qui spécifie les méthodes standard que tout autre référentiel doit implémenter (AbstractRepository). Et puis, un référentiel particulier est défini avec l'implémentation concrète de ces méthodes où la logique d'accès aux données est implémentée (par exemple, SQLAlchemyRepository). Ce modèle de conception vise à isoler les méthodes de manipulation des données afin qu'elles puissent être utilisées de manière transparente ailleurs dans l'application, par ex. dans notre modèle de domaine.

Modèle d'unité de travail

Le modèle d'unité de travail est la pièce manquante pour enfin découpler le modèle de l'accès aux données. L'unité de travail encapsule la logique d'accès aux données et permet de regrouper toutes les opérations qui doivent être effectuées sur la source de données au sein d'une seule transaction. Ce modèle garantit que toutes les opérations sont effectuées de manière atomique.

Lors de l'implémentation du modèle d'unité de travail, nous définissons généralement une interface qui spécifie les méthodes standard que toute autre unité de travail doit implémenter (AbstractUnitOfWork). Et puis, une unité de travail particulière est définie avec l'implémentation concrète de ces méthodes où la logique d'accès aux données est implémentée (par exemple, SQLAlchemyUnitOfWork). Cette conception permet une gestion systématique de la connexion à la source de données, sans qu'il soit nécessaire de modifier la mise en œuvre de la logique métier de l'application.

Implémentation de DDD avec Flama

Après la rapide introduction aux principaux concepts de DDD, nous sommes prêts à plonger dans la mise en œuvre de DDD avec Flama. Dans cette section, nous vous guiderons tout au long du processus de configuration de l'environnement de développement, de création d'une application de base et de mise en œuvre des concepts DDD avec Flama.

Avant de continuer avec l'exemple, veuillez jeter un œil à la convention de dénomination de Flama concernant les principaux concepts DDD que nous venons de passer en revue :

Comme vous pouvez le voir dans la figure ci-dessus, la convention de dénomination est assez intuitive : Repository fait référence au modèle de référentiel ; et Travailleur fait référence à l'unité de travail. Maintenant, nous pouvons maintenant passer à la mise en œuvre d'une API Flama qui utilise DDD. Mais, avant de commencer, si vous avez besoin de revoir les bases sur la façon de créer une API simple avec flama, ou sur la façon d'exécuter l'API une fois que le code est déjà prêt, alors vous voudrez peut-être vérifier consultez le guide de démarrage rapide. Vous y trouverez les concepts fondamentaux et les étapes nécessaires pour suivre cet article. Maintenant, sans plus tarder, commençons par la mise en œuvre.

Mise en place de l'environnement de développement

Notre première étape consiste à créer notre environnement de développement et à installer toutes les dépendances requises pour ce projet. La bonne nouvelle est que pour cet exemple, il suffit d'installer flama pour disposer de tous les outils nécessaires à la mise en œuvre de l'authentification JWT. Nous utiliserons la poésie pour gérer nos dépendances, mais vous pouvez également utiliser pip si vous préférez :

poetry add "flama[full]" "aiosqlite"

Le package aiosqlite est requis pour utiliser SQLite avec SQLAlchemy, qui est la base de données que nous utiliserons dans cet exemple.

Si vous voulez savoir comment nous organisons généralement nos projets, jetez un œil à notre article précédent ici, où nous expliquons en détail comment mettre en place un projet python avec de la poésie et la structure des dossiers de projet que nous suivons habituellement.

Application de base

Commençons par une application simple dotée d'un seul point de terminaison public. Ce point de terminaison renverra une brève description de l'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}

Si vous souhaitez exécuter cette application, vous pouvez enregistrer le code ci-dessus dans un fichier appelé app.py sous le dossier src, puis exécuter la commande suivante (n'oubliez pas d'activer l'environnement de poésie, sinon vous devrez préfixez la commande avec poésie 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)

où l'indicateur --server-reload est facultatif et est utilisé pour recharger automatiquement le serveur lorsque le code change. Ceci est très utile lors du développement, mais vous pouvez le supprimer si vous n'en avez pas besoin. Pour une liste complète des options disponibles, vous pouvez exécuter flama run --help ou consulter la documentation.

Vous pouvez également exécuter l'application en exécutant le script suivant, que vous pouvez enregistrer sous __main__.py dans le dossier src :

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

Et puis, vous pouvez exécuter l'application en exécutant la commande suivante :

poetry add "flama[full]" "aiosqlite"

DDD en action

Maintenant, après avoir mis en place un squelette minimal pour notre application, nous pouvons commencer à mettre en œuvre les concepts DDD que nous venons de passer en revue dans le
contexte d’un exemple simple qui tente d’imiter un scénario du monde réel. Supposons que l'on nous demande de développer une API pour gérer les utilisateurs et que nous disposons des exigences suivantes :

• Nous souhaitons créer de nouveaux utilisateurs via une requête POST à ​​/user/, en fournissant le nom, le prénom, l'e-mail et le mot de passe de l'utilisateur.
• Tout utilisateur créé sera stocké dans une base de données avec le schéma suivant :
  • id : identifiant unique de l'utilisateur.
  • nom : nom de l'utilisateur.
  • nom : nom de famille de l'utilisateur.
  • email : email de l'utilisateur.
  • mot de passe : mot de passe de l'utilisateur. Cela doit être haché avant de le stocker dans la base de données.
  • actif : un indicateur booléen pour indiquer si l'utilisateur est actif ou non. Par défaut, les utilisateurs sont créés comme inactifs.
• Les utilisateurs créés doivent activer leur compte en envoyant une requête POST à ​​/user/activate/ avec leur email et leur mot de passe. Une fois l'utilisateur activé, le statut de l'utilisateur doit être mis à jour dans la base de données comme actif.
• Les utilisateurs peuvent se connecter en envoyant une requête POST à ​​/user/signin/ avec leur e-mail et leur mot de passe. Si l'utilisateur est actif, l'API doit renvoyer toutes les informations de l'utilisateur. Sinon, l'API doit renvoyer un message d'erreur.
• Les utilisateurs qui souhaitent désactiver leur compte peuvent le faire en envoyant une requête POST à ​​/user/deactivate/ avec leur email et leur mot de passe. Une fois l'utilisateur désactivé, le statut de l'utilisateur doit être mis à jour dans la base de données comme inactif.

Cet ensemble d'exigences constitue ce que nous avons précédemment appelé le modèle de domaine de notre application, qui n'est essentiellement qu'une matérialisation du flux de travail utilisateur suivant :

1. Un utilisateur est créé via une requête POST à ​​/user/.
2. L'utilisateur active son compte via une requête POST à ​​/user/activate/.
3. L'utilisateur se connecte via une requête POST à ​​/user/signin/.
4. L'utilisateur désactive son compte via une requête POST à ​​/user/deactivate/.
5. L'utilisateur peut répéter les étapes 2 à 4 autant de fois qu'il le souhaite.

Maintenant, implémentons le modèle de domaine en utilisant les modèles de référentiel et de travail. Nous commencerons par définir le modèle de données, puis nous implémenterons les modèles de référentiel et de travail.

Modèle de données

Les données de nos utilisateurs seront stockées dans une base de données SQLite (vous pouvez utiliser n'importe quelle autre base de données prise en charge par SQLAlchemy). Nous utiliserons le modèle de données suivant pour représenter les utilisateurs (vous pouvez enregistrer ce code dans un fichier appelé models.py sous le dossier src) :

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

Outre le modèle de données, nous avons besoin d'un script de migration pour créer la base de données et la table. Pour cela, nous pouvons enregistrer le code suivant dans un fichier appelé migrations.py à la racine du projet :

poetry add "flama[full]" "aiosqlite"

Et puis, nous pouvons exécuter le script de migration en exécutant la commande suivante :

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

Dépôt

Dans cet exemple nous n'aurons besoin que d'un seul dépôt, à savoir le dépôt qui gérera les opérations atomiques sur la table utilisateur, dont le nom sera UserRepository. Heureusement, flama fournit une classe de base pour les référentiels liés aux tables SQLAlchemy, appelée SQLAlchemyTableRepository.

La classe SQLAlchemyTableRepository fournit un ensemble de méthodes pour effectuer des opérations CRUD sur la table, notamment :

• create : crée de nouveaux éléments dans le tableau. Si l'élément existe déjà, il déclenchera une exception (IntegrityError), sinon il renverra la clé primaire du nouvel élément.
• retrieve : Récupère un élément de la table. Si l'élément n'existe pas, il déclenchera une exception (NotFoundError), sinon il renverra l'élément. Si plusieurs éléments sont trouvés, une exception sera déclenchée (MultipleRecordsError).
• update : Met à jour un élément dans le tableau. Si l'élément n'existe pas, il déclenchera une exception (NotFoundError), sinon il renverra l'élément mis à jour.
• delete : Supprime un élément de la table.
• list : Répertorie tous les éléments du tableau qui correspondent aux clauses et filtres passés. Si aucune clause ou filtre n'est fourni, il renvoie tous les éléments du tableau. Si aucun élément n'est trouvé, il renvoie une liste vide.
• drop : supprime la table de la base de données.

Pour les besoins de notre exemple, nous n'avons besoin d'aucune action supplémentaire sur la table, les méthodes fournies par SQLAlchemyTableRepository sont donc suffisantes. Nous pouvons enregistrer le code suivant dans un fichier appelé repositories.py sous le dossier src :

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)

Comme vous pouvez le voir, la classe UserRepository est une sous-classe de SQLAlchemyTableRepository, et elle nécessite uniquement que la table soit définie dans l'attribut _table. C'est la seule chose que nous devons faire pour avoir un référentiel entièrement fonctionnel pour la table utilisateur.

Si nous voulions ajouter des méthodes personnalisées au-delà des opérations CRUD standard, nous pourrions le faire en les définissant dans la classe UserRepository. Par exemple, si nous voulions ajouter une méthode pour compter le nombre d'utilisateurs actifs, nous pourrions le faire comme suit :

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

Bien que nous n'utiliserons pas cette méthode dans notre exemple, il est bon de savoir que nous pouvons ajouter des méthodes personnalisées au référentiel si nécessaire, et comment elles sont implémentées
dans le contexte du modèle de référentiel. Il s'agit d'un modèle de conception puissant comme nous pouvons déjà le constater, puisque nous pouvons implémenter ici toute la logique d'accès aux données sans avoir à modifier la logique métier de l'application (qui est implémentée dans les méthodes de ressources correspondantes).

Travailleur

Le modèle d'unité de travail est utilisé pour encapsuler la logique d'accès aux données et fournir un moyen de regrouper toutes les opérations qui doivent être effectuées sur la source de données au sein d'une seule transaction. Dans flama le modèle UoW est implémenté avec le nom de Worker. De la même manière que pour le modèle de référentiel, flama fournit une classe de base pour les travailleurs liés aux tables SQLAlchemy, appelée SQLAlchemyWorker. Essentiellement, SQLAlchemyWorker fournit une connexion et une transaction à la base de données, et instancie tous ses référentiels avec la connexion du travailleur. Dans cet exemple, notre travailleur n'utilisera qu'un seul référentiel (à savoir le UserRepository) mais nous pourrions ajouter plus de référentiels si nécessaire.

Notre travailleur s'appellera RegisterWorker, et nous pouvons enregistrer le code suivant dans un fichier appelé Workers.py sous le dossier src :

poetry add "flama[full]" "aiosqlite"

Ainsi, si nous avions plus de référentiels avec lesquels travailler, par exemple ProductRepository et OrderRepository, nous pourrions les ajouter au travailleur comme suit :

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

Aussi simple que cela, nous avons implémenté les modèles de référentiel et de travail dans notre application. Nous pouvons maintenant passer à la mise en œuvre des méthodes de ressources qui fourniront les points de terminaison d'API nécessaires pour interagir avec les données utilisateur.

Ressources

Les ressources sont l'un des principaux éléments constitutifs d'une application flama. Ils sont utilisés pour représenter les ressources applicatives (au sens de ressources RESTful) et pour définir les points de terminaison de l'API qui interagissent avec elles.

Dans notre exemple, nous définirons une ressource pour l'utilisateur, appelée UserResource, qui contiendra les méthodes pour créer, activer, connecter et désactiver des utilisateurs. Les ressources doivent dériver, au moins, de la classe Resource intégrée flama, bien que flama fournisse des classes plus sophistiquées avec lesquelles travailler, telles que RESTResource et CRUDResource.

Nous pouvons enregistrer le code suivant dans un fichier appelé resources.py sous le dossier src :

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)

Application de base avec DDD

Maintenant que nous avons implémenté le modèle de données, les modèles de référentiel et de travail, ainsi que les méthodes de ressources, nous devons modifier l'application de base que nous avons introduite précédemment, afin que tout fonctionne comme prévu. Nous devons :

• Ajoutez la connexion SQLAlchemy à l'application, et ceci est réalisé en ajoutant le SQLAlchemyModule au constructeur de l'application en tant que module.
• Ajoutez le travailleur à l'application, et cela est réalisé en ajoutant RegisterWorker au constructeur de l'application en tant que composant.

Cela laissera le fichier app.py comme suit :

poetry add "flama[full]" "aiosqlite"

Vous devriez déjà comprendre comment le modèle DDD nous a permis de séparer la logique métier de l'application (qui est facilement lisible dans les méthodes de ressources) de la logique d'accès aux données (qui est implémenté dans le référentiel et les modèles de travail). Il convient également de noter comment cette séparation des préoccupations a rendu le code plus maintenable et testable, et comment le code est désormais mieux aligné sur les exigences métier qui nous ont été données au début de cet exemple.

Exécuter l'application

Avant d'exécuter une commande, veuillez vérifier que votre environnement de développement est correctement configuré et que la structure des dossiers est la suivante :

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

Si tout est correctement configuré, vous pouvez exécuter l'application en exécutant la commande suivante (pensez à exécuter le script de migration avant d'exécuter l'application) :

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)

Nous pouvons maintenant essayer la logique métier que nous venons de mettre en œuvre. N'oubliez pas que vous pouvez essayer cela soit en utilisant un outil comme curl ou Postman, soit en utilisant l'interface utilisateur des documents générés automatiquement fournis par flama en accédant à http://localhost:8000/docs/ dans votre navigateur. et essayer les points de terminaison à partir de là.

Créer un utilisateur

Pour créer un utilisateur, vous pouvez envoyer une requête POST à ​​/user/ avec la charge utile suivante :

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

Nous pouvons donc utiliser curl pour envoyer la requête comme suit :

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)

Si la demande aboutit, vous devriez recevoir une réponse 200 avec un corps vide et l'utilisateur sera créé dans la base de données.

Se connecter

Pour vous connecter, vous pouvez envoyer une requête POST à ​​/user/signin/ avec la charge utile suivante :

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

Nous pouvons donc utiliser curl pour envoyer la requête comme suit :

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

Étant donné que l'utilisateur n'est pas actif, vous devriez recevoir quelque chose comme la réponse suivante :

> poetry run python migrations.py

Database and User table created successfully.

Nous pouvons également tester ce qui se passerait si quelqu'un essayait de se connecter avec un mauvais mot de passe :

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

from src import models

__all__ = ["UserRepository"]

class UserRepository(SQLAlchemyTableRepository):
    _table = models.user_table

Dans ce cas, vous devriez recevoir une réponse 401 avec le corps suivant :

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

Enfin, nous devrions également essayer de nous connecter avec un utilisateur qui n'existe pas :

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

from src import repositories

__all__ = ["RegisterWorker"]


class RegisterWorker(SQLAlchemyWorker):
    user: repositories.UserRepository

Dans ce cas, vous devriez recevoir une réponse 404 avec le corps suivant :

# 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

Activation de l'utilisateur

Après avoir exploré le processus de connexion, nous pouvons maintenant activer l'utilisateur en envoyant une requête POST à ​​/user/activate/ avec les informations d'identification de l'utilisateur :

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

Avec cette demande, l'utilisateur doit être activé et vous devriez recevoir une réponse 200 avec un corps vide.

Comme dans le cas précédent, nous pouvons également tester ce qui se passerait si quelqu'un tentait d'activer l'utilisateur avec un mauvais mot de passe :

poetry add "flama[full]" "aiosqlite"

Dans ce cas, vous devriez recevoir une réponse 401 avec le corps suivant :

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

Enfin, il faudrait aussi essayer d'activer un utilisateur qui n'existe pas :

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)

Dans ce cas, vous devriez recevoir une réponse 404 avec le corps suivant :

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

Connexion de l'utilisateur après l'activation

Maintenant que l'utilisateur est activé, nous pouvons essayer de nous connecter à nouveau :

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)

Ce qui, cette fois, devrait renvoyer une réponse de 200 avec les informations de l'utilisateur :

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

Désactivation de l'utilisateur

Enfin, nous pouvons désactiver l'utilisateur en envoyant une requête POST à ​​/user/deactivate/ avec les identifiants de l'utilisateur :

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

Avec cette demande, l'utilisateur doit être désactivé et vous devriez recevoir une réponse 200 avec un corps vide.

Conclusion

Dans cet article, nous nous sommes aventurés dans le monde du Domain-Driven Design (DDD) et comment il peut être implémenté dans une application flama. Nous avons vu comment DDD peut nous aider à séparer la logique métier de l'application de la logique d'accès aux données, et comment cette séparation des préoccupations peut rendre le code plus maintenable et testable. Nous avons également vu comment les modèles de référentiel et de travail peuvent être implémentés dans une application flama, et comment ils peuvent être utilisés pour encapsuler la logique d'accès aux données et fournir un moyen de regrouper toutes les opérations qui doivent être effectuées. sur la source de données en une seule transaction. Enfin, nous avons vu comment les méthodes de ressources peuvent être utilisées pour définir les points de terminaison de l'API qui interagissent avec les données utilisateur, et comment le modèle DDD peut être utilisé pour implémenter les exigences métier qui nous ont été données au début de cet exemple.

Bien que le processus de connexion que nous avons décrit ici ne soit pas entièrement réaliste, vous pouvez combiner ce contenu et un article précédent sur l'authentification JWT pour implémenter un processus plus réaliste, dans lequel la connexion finit par renvoyer un Jeton JWT. Si cela vous intéresse, vous pouvez consulter l'article sur l'authentification JWT avec flama.

Nous espérons que vous avez trouvé cet article utile et que vous êtes maintenant prêt à implémenter DDD dans vos propres applications flama. Si vous avez des questions ou des commentaires, n'hésitez pas à nous contacter. Nous sommes toujours heureux de vous aider !

Restez à l'écoute pour plus d'articles sur flama et d'autres sujets passionnants dans le monde de l'IA et du développement de logiciels. À la prochaine fois !

Soutenez notre travail

Si vous aimez ce que nous faisons, il existe un moyen simple et gratuit de soutenir notre travail. Offrez-nous un ⭐ chez Flama.

GitHub ⭐ représente un monde pour nous et nous donne le carburant le plus doux pour continuer à travailler dessus et aider les autres dans son cheminement vers la création d'API d'apprentissage automatique robustes.

Vous pouvez également nous suivre sur ?, où nous partageons nos dernières nouvelles et mises à jour, ainsi que des discussions intéressantes sur l'IA, le développement de logiciels et bien plus encore.

Références

• Documentation Flama
• Dépôt Flama GitHub
• Pack Flama PyPI

À propos des auteurs

• Vortico : Nous sommes spécialisés dans le développement de logiciels pour aider les entreprises à améliorer et à étendre leurs capacités en matière d'IA et de technologie.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!