Créez une base de données propre pour les tests avec FastAPI et effectuez le test Unittest de l'API avec pytest

Choses à faire

  1. [Implémentation d'une application CRUD simple à l'aide de sqlite3 avec FastAPI](# Implémentation d'une application CRUD simple à l'aide de sqlite3 avec FastAPI)
  2. [Créez une base de données pour chaque cas de test et effectuez un test unitaire API avec pytest sans affecter les autres cas de test ou la base de données de production](#Créez une base de données pour chaque cas de test et autres cas de test Et unittest l'API avec pytest sans affecter la base de données de production)

1. Implémentation d'une application CRUD simple à l'aide de sqlite3 avec FastAPI

Application CRUD

Créez une application CRUD pour votre compte utilisateur. Puisque l'introduction de la méthode de test est la principale, seules les fonctions simples suivantes seront implémentées.

--Créer: enregistrement de l'utilisateur

Préparation

En plus de FastAPI, les packages suivants de pip install sont requis.

Structure du répertoire

Mettez en œuvre ce dont vous avez besoin avec la structure de répertoires suivante.

users
├── __init__.py
├── crud.py          #Définition de fonction pour l'émission de requêtes
├── conftest.py      #définition du luminaire pytest
├── database.py      #Paramètres de la base de données
├── main.py          #Définition de l'API
├── models.py        #définition de table
├── schemas.py       #API I/O définition
└── tests
    ├── __init__.py
    └── test_user.py #Test API

Paramètres de la base de données (database.py)

Spécifiez la base de données à connecter avec l'URL de la base de données. Fondamentalement, elle doit être déclarée comme une variable d'environnement, mais pour des raisons de simplicité, elle est écrite de manière solide. La notation des URL des principales bases de données est résumée dans ici.

False}Étant donné que la partie de est le paramètre de sqlite3, veuillez le supprimer lorsque vous utilisez une autre base de données.



```variable sessionlocal```Est dans```sessionmaker```instanse est,
 L'appel créera une instance de session. Ceci est utilisé pour gérer la connexion avec le DB. Nous utilisons également session pour émettre des requêtes SQL.


#### **`database.py`**
```python

import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = os.environ.get('DATABASE_URL', 'sqlite:///./test.db')

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Définition de la table (models.py)

La table est définie en héritant de la base définie lors de la configuration de la base de données. Avec cette définition, vous pouvez facilement créer une table et utiliser le mappeur ORM via Base.

models.py


from sqlalchemy import Boolean, Column, Integer, String
from .database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

Définition de fonction pour l'émission de requêtes (crud.py)

sqlalchemy utilise session pour émettre des requêtes SQL. Ce processus est sujet à des problèmes, alors supprimez-le pour faciliter les tests unitaires. Autant que possible, je n'émettrai une requête qu'en recevant une session sans insérer de logique.

crud.py


from sqlalchemy.orm import Session
from hashlib import md5 as hash_func
from . import models

def get_user_by_email_query(db: Session, email: str):
    """get user by email"""
    return db.query(models.User).filter(models.User.email == email).first()

def create_user_query(db: Session, email: str, password: str):
    """create user by email and password"""
    hashed_password = hash_func(password.encode()).hexdigest()
    db_user = models.User(email=email, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Définition d'E / S API (schemas.py)

Définit les E / S API. Ici, entrez l'email et le mot de passe-> renvoyez l'id, l'email et l'utilisateur actif de l'utilisateur créé. Décidez simplement du schéma et il effectuera la sérialisation et la désérialisation sans autorisation.

schemas.py


from pydantic import BaseModel

class UserBase(BaseModel):
    """Base User scheme"""
    email: str

class UserCreate(UserBase):
    """Input"""
    password: str

class User(UserBase):
    """Output"""
    id: int
    is_active: bool

    class Config:
        orm_mode = True

Définition de l'API (main.py)

Définissez l'API CRUD. Une chose à savoir est de savoir comment réussir la session. Si vous déclarez une fonction ou une classe comme `` dépend '' dans l'argument, le résultat de son appel (retour pour fonction, instance pour classe) est passé à l'argument. Grâce à cela, une session est créée à l'aide de SessionLocal pour chaque demande et une connexion avec la base de données est sécurisée. Ensuite, la requête est émise à l'aide de cette session.

main.py


from typing import List
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from . import models, schemas
from .crud import (
    get_user_by_email_query,
    create_user_query
)
from .database import SessionLocal, engine

#création de table
models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency
def get_db():
    try:
        db = SessionLocal() #Générer une session
        yield db
    finally:
        db.close()

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = get_user_by_email_query(db=db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return create_user_query(db=db, user=user)

Il existe également une implémentation qui utilise un middleware comme méthode de transmission d'une session, mais comme toutes les API créent une connexion avec la base de données, il y a des effets néfastes tels que le gaspillage si de nombreuses API n'utilisent pas la base de données. Semble être obsolète. (Référence)

2. Créez un DB pour chaque scénario de test et effectuez un test unitaire de l'API sans affecter les autres scénarios de test ou le DB de production.

Commencera le sujet principal.

Test d'API (lors de l'utilisation d'une base de données de production)

Dans FastAPI, vous pouvez simplement tester l'API avec starlette.testclient.TestClient '' comme suit.

test_user.py


from starlette.testclient import TestClient
from users.main import app

client = TestClient(app)

def test_create_user():
    response = client.post(
        "/users/", json={"email": "foo", "password": "fo"}
    )
    assert response.status_code == 200

Maintenant, exécutez pytest pour des tests automatisés.

$ pytest

Cependant, l'API se connecte à la base de données de production, donc lorsque j'exécute le test, j'ajoute un utilisateur. Si vous exécutez le test deux fois, l'e-mail portant le même nom est déjà enregistré la deuxième fois, de sorte que la création de l'utilisateur échoue et le test échoue.

Par conséquent, créez une base de données temporairement au moment de l'exécution du test afin que la base de données pour la production ne soit pas affectée et qu'une base de données propre puisse être préparée à chaque exécution du test. De plus, nous recréerons la base de données pour chaque fonction afin qu'elle puisse être utilisée à des fins générales afin qu'elle ne s'affecte pas pour chaque cas de test.

Créer et supprimer une base de données propre pour le test

Le traitement requis pour tester avec un DB propre est le suivant.

Si vous pouvez faire ces choses, vous pouvez utiliser une base de données propre pour chaque cas de test, ne laissant aucune trace à la fin. Ce processus est requis quel que soit le cas de test, nous allons donc définir un appareil pour effectuer un tel processus dans conftest.py```. La mise en œuvre ressemble à ceci:

conftest.py


import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import database_exists, drop_database
from .database import Base

@pytest.fixture(scope="function")
def SessionLocal():
    # settings of test database
    TEST_SQLALCHEMY_DATABASE_URL = "sqlite:///./test_temp.db"
    engine = create_engine(TEST_SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})

    assert not database_exists(TEST_SQLALCHEMY_DATABASE_URL), "Test database already exists. Aborting tests."

    # Create test database and tables
    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    # Run the tests
    yield SessionLocal

    # Drop the test database
    drop_database(TEST_SQLALCHEMY_DATABASE_URL)

Modifier la base de données à laquelle l'API se connecte pendant le test

En déclarant SessionLocal '' dans l'argument grâce au fixture, une base de données propre est maintenant créée lorsque la fonction est exécutée. Après cela, il est nécessaire de changer de force la base de données à laquelle l'API se connecte à celle à tester. Je veux le compléter avec juste le code de test pour réduire l'impact. Dans FastAPI, le FastAPI.Depends``` déclaré dans l'argument API peut être écrasé de force par ```app.dependency_overrides```. Ainsi, vous pouvez changer la destination de la connexion en écrasant main.get_db '' et en la réécrivant pour utiliser l'instance Sessionmaker pour les tests. Par conséquent, définissez le décorateur suivant.

test_user.py


from users.main import app, get_db

def temp_db(f):
    def func(SessionLocal, *args, **kwargs):
        #Sessionmaker instanse pour se connecter à la base de données de test
        #  (SessionLocal)Du luminaire

        def override_get_db():
            try:
                db = SessionLocal()
                yield db
            finally:
                db.close()

        #utiliser SessionLocal reçu de l'appareil_Forcer le changement de base de données
        app.dependency_overrides[get_db] = override_get_db
        # Run tests
        f(*args, **kwargs)
        # get_Annuler la base de données
        app.dependency_overrides[get_db] = get_db
    return func

En modifiant simplement le code de test et en utilisant le décorateur défini précédemment, vous pouvez créer une base de données temporaire à tester au moment de l'exécution du test, fonction par fonction, et tester en utilisant cette base de données. De plus, comme chacun utilisera une base de données indépendante, cela n'affectera pas les autres cas de test.

test_user.py


from starlette.testclient import TestClient
from users.main import app

client = TestClient(app)

@temp_db
def test_create_user():
    response = client.post(
        "/users/", json={"email": "foo", "password": "fo"}
    )
    assert response.status_code == 200

en conclusion

J'ai résumé comment créer une base de données de test avec FastAPI et effectuer Unittest d'API avec pytest. Recréer la base de données pour chaque cas de test semble ralentir la vitesse de traitement, j'ai donc essayé et erroné la méthode de restauration, mais j'ai abandonné. Je pense que cette méthode est également efficace lorsqu'il y a peu de cas de test ou que la quantité de données lors des tests n'est pas importante, j'espère donc que cet article vous sera utile!

Refs

Recommended Posts

Créez une base de données propre pour les tests avec FastAPI et effectuez le test Unittest de l'API avec pytest
Créez un modèle pour stocker les informations de l'API Google Livres pour une manipulation et des tests intuitifs
Créez une carte thermique de tweet avec l'API Google Maps
Créez un Twitter BOT avec le SDK GoogleAppEngine pour Python
Créez une API d'intégration sociale pour les applications smartphone avec Django
Analysez l'API Researchmap avec Python et créez automatiquement un fichier Word pour la liste des succès
Obtenez le prix d'achat et de vente de la monnaie virtuelle avec l'API de l'échange Zaif et créez un graphique
Créer une API CRUD à l'aide de l'API rapide
Créez un alias pour Route53 vers CloudFront avec l'API AWS
Créez une illusion rayée avec correction gamma pour Python3 et openCV3
Créez un sélecteur de couleurs pour la roue chromatique avec Python + Qt (PySide)
Créer une API REST pour faire fonctionner dynamodb avec le Framework Django REST
Créer et renvoyer un fichier CSV CP932 pour Excel avec Chalice
Peut être fait en 5 minutes!? Créez une API de détection de visage avec Fast API et OpenCV et publiez-la sur Heroku
Créez une commande pour rechercher des composés similaires dans la base de données cible avec RDKit et vérifiez le temps de traitement
Créez une application Web typée avec le framework Web de Python «Fast API» et TypeScript / Open API - Pile technique pour les applications Web d'apprentissage automatique
Créez une API pour convertir des fichiers PDF en images TIF avec FastAPI et Docker
Créez une API REST à l'aide du modèle appris dans Lobe et TensorFlow Serving.
Créons un système de réception simple avec le framework sans serveur Python Chalice et Twilio
Utilisez la commande [shell] pour compresser par zip n'importe quel fichier pour créer un fichier et supprimer le fichier d'origine.
Essayez la touche d'un test basé sur les données avec Selenium Python Bindings et py.test
Création d'un wrapper Python pour l'API Qiita
Créer un LINE BOT avec Minette pour Python
Créez un tableau de bord pour les appareils réseau avec Django!
Créez un outil de traduction avec Translate Toolkit
Développer une API Web qui renvoie les données stockées dans DB avec Django et SQLite
Construisez un serveur API pour vérifier le fonctionnement de l'implémentation frontale avec python3 et Flask
Créez une application graphique native avec Py2app et Tkinter
Comment créer un sous-menu avec le plug-in [Blender]
Créez un code QR pour l'URL sous Linux
[Boto3] Rechercher des utilisateurs Cognito avec l'API List Users
Créez un lot d'images et gonflez avec ImageDataGenerator
Obtenez des commentaires et des abonnés avec l'API de données YouTube
Créer une couche pour AWS Lambda Python dans Docker
Créez un fichier audio avec la fonction de synthèse vocale de Google Text To Speak et vérifiez le texte comme guide de parole pendant 3 minutes.
[Linux] Créez un auto-certificat avec Docker et apache
Un script qui facilite la création de menus riches avec l'API de messagerie LINE
Créez un script pour votre compétence Pepper dans une feuille de calcul et chargez SayText directement à partir du script
Essayez d'utiliser l'API Twitter rapidement et facilement avec Python
Créer un nouveau csv avec des pandas basé sur le csv local
Une note sur l'utilisation de l'API Facebook avec le SDK Python
Créez une caméra de surveillance WEB avec Raspberry Pi et OpenCV
Créer un compte enfant de connect with Stripe en Python
Probablement le moyen le plus simple de créer un pdf avec Python 3
[Python] Créer une liste de dates et d'heures pour une période spécifiée
Créez des applications, enregistrez des données et partagez-les avec un seul e-mail
Créons un diagramme PRML avec Python, Numpy et matplotlib.
Procédure d'installation pour Python et Ansible avec une version spécifique
Créez un robot de réponse automatique en temps réel à l'aide de l'API Twitter Streaming
Créez une partition, puis installez le système d'exploitation Raspberry Pi
Rationalisez la collecte d'informations avec l'API Twitter et les robots Slack
Tornado - Créons une API Web qui renvoie facilement JSON avec JSON
Créez un script de déploiement avec fabric et cuisine et réutilisez-le
Créez une API Web capable de fournir des images avec Django
Fabriquez un thermomètre BLE et obtenez la température avec Pythonista3
Je veux créer un Dockerfile pour le moment.
[Python] Créer un écran pour le code d'état HTTP 403/404/500 avec Django
J'ai essayé d'utiliser l'API Google avec Ruby et Python - Faites de la base de données une feuille de calcul et gérez-la avec Google Drive
J'ai essayé de créer un traitement par lots sans serveur pour la première fois avec DynamoDB et Step Functions
Comprendre les probabilités et les statistiques qui peuvent être utilisées pour la gestion des progrès avec un programme python