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
En plus de FastAPI, les packages suivants de pip install sont requis.
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
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()
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)
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é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é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)
Commencera le sujet principal.
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.
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)
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
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