Erstellen Sie eine CRUD-App für Ihr Benutzerkonto. Da die Einführung der Testmethode die wichtigste ist, werden nur die folgenden einfachen Funktionen implementiert.
--Erstellen: Benutzerregistrierung
Zusätzlich zu FastAPI sind die folgenden Pakete für die Pip-Installation erforderlich.
Implementieren Sie das, was Sie brauchen, mit der folgenden Verzeichnisstruktur.
users
├── __init__.py
├── crud.py #Funktionsdefinition für die Ausgabe von Abfragen
├── conftest.py #pytest Fixture Definition
├── database.py #Datenbankeinstellungen
├── main.py #API-Definition
├── models.py #Tabellendefinition
├── schemas.py #API I./O Definition
└── tests
├── __init__.py
└── test_user.py #API-Tests
Geben Sie die Datenbank an, die mit der Datenbank-URL verbunden werden soll. Grundsätzlich sollte es als Umgebungsvariable deklariert werden, aber der Einfachheit halber ist es solide geschrieben. Die Notation der URLs der wichtigsten Datenbanken ist in [hier] zusammengefasst (https://docs.sqlalchemy.org/en/13/core/engines.html).
False}Da der Teil von die Einstellung für sqlite3 ist, löschen Sie sie bitte, wenn Sie eine andere Datenbank verwenden.
```sitzungslokale Variable```Ist in```sessionmaker```Instanse ist,
Der Aufruf erstellt eine Sitzungsinstanz. Dies wird verwendet, um die Verbindung mit der DB zu verwalten. Wir verwenden die Sitzung auch zum Ausgeben von SQL-Abfragen.
#### **`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()
Die Tabelle wird definiert, indem die beim Festlegen der Datenbank definierte Basis geerbt wird. Mit dieser Definition können Sie einfach eine Tabelle erstellen und den ORM-Mapper über Base verwenden.
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 verwendet eine Sitzung, um SQL-Abfragen auszugeben. Dieser Prozess ist anfällig für Probleme. Schneiden Sie ihn daher aus, um die Durchführung von Komponententests zu vereinfachen. So viel wie möglich werde ich eine Abfrage nur ausgeben, indem ich eine Sitzung empfange, ohne Logik einzufügen.
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
Definiert API I / O. Geben Sie hier E-Mail und Passwort ein -> geben Sie die ID, die E-Mail-Adresse und den aktiven Benutzer des erstellten Benutzers zurück. Entscheiden Sie einfach das Schema und es wird Serialize und Deserialize ohne Erlaubnis ausführen.
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
Definieren Sie die CRUD-API.
Eine Sache, die Sie beachten sollten, ist, wie Sie die Sitzung bestehen. Wenn Sie eine Funktion oder Klasse im Argument als `abhängig `deklarieren, wird das Ergebnis des Aufrufs (Rückgabe für Funktion, Instanz für Klasse) an das Argument übergeben. Auf diese Weise wird mit SessionLocal für jede Anforderung eine Sitzung erstellt und eine Verbindung mit der Datenbank sichergestellt. Dann wird die Abfrage unter Verwendung dieser Sitzung ausgegeben.
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
#Tabellenerstellung
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
# Dependency
def get_db():
try:
db = SessionLocal() #Sitzung generieren
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)
Es gibt auch eine Implementierung, die Middleware als Methode zum Übergeben einer Sitzung verwendet. Da jedoch alle APIs eine Verbindung mit der Datenbank herstellen, gibt es schädliche Auswirkungen wie Verschwendung, wenn viele APIs die Datenbank nicht verwenden. Scheint veraltet zu sein. (Referenz)
Beginnt das Hauptthema.
In FastAPI können Sie die API einfach wie folgt mit `` `starlette.testclient.TestClient``` testen.
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
Führen Sie jetzt pytest für automatisierte Tests aus.
$ pytest
Die API stellt jedoch eine Verbindung zur Produktionsdatenbank her. Wenn ich den Test ausführe, füge ich einen Benutzer hinzu. Wenn Sie den Test zweimal ausführen, wird die gleichnamige E-Mail bereits zum zweiten Mal registriert, sodass die Benutzererstellung fehlschlägt und der Test nicht bestanden wird.
Erstellen Sie daher vorübergehend eine Datenbank zum Zeitpunkt der Testausführung, damit die Produktionsdatenbank nicht betroffen ist, und bei jeder Testausführung kann eine saubere Datenbank erstellt werden. Darüber hinaus erstellen wir die Datenbank für jede Funktion neu, damit sie für allgemeine Zwecke verwendet werden kann, damit sie sich nicht für jeden Testfall gegenseitig beeinflusst.
Die zum Testen mit einer sauberen Datenbank erforderliche Verarbeitung ist wie folgt.
Wenn Sie diese Dinge tun können, können Sie für jeden Testfall eine saubere Datenbank verwenden, die am Ende keine Spuren hinterlässt. Dieser Prozess ist unabhängig vom Testfall erforderlich, daher definieren wir in `` `conftest.py``` ein Gerät, um einen solchen Prozess durchzuführen. Die Implementierung sieht folgendermaßen aus:
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)
Durch Deklarieren von "SessionLocal" im Argument dank des Scheinwerfers wird jetzt eine saubere Datenbank erstellt, wenn die Funktion ausgeführt wird. Danach muss die Datenbank, mit der die API eine Verbindung zum Test herstellt, zwangsweise geändert werden. Ich möchte es nur mit dem Testcode vervollständigen, um die Auswirkungen zu verringern.
In FastAPI kann das im API-Argument deklarierte `FastAPI.Depends``` mit` app.dependency_overrides zwangsweise überschrieben werden. Sie können das Verbindungsziel also ändern, indem Sie `` `main.get_db überschreiben und neu schreiben, um die Sessionmaker-Instanz zum Testen zu verwenden.
Definieren Sie daher den folgenden Dekorator.
test_user.py
from users.main import app, get_db
def temp_db(f):
def func(SessionLocal, *args, **kwargs):
#Sessionmaker instanse, um eine Verbindung zur Test-DB herzustellen
# (SessionLocal)Vom Gerät
def override_get_db():
try:
db = SessionLocal()
yield db
finally:
db.close()
#Verwenden Sie SessionLocal, das vom Gerät empfangen wurde_Datenbankwechsel erzwingen
app.dependency_overrides[get_db] = override_get_db
# Run tests
f(*args, **kwargs)
# get_Db rückgängig machen
app.dependency_overrides[get_db] = get_db
return func
Durch einfaches Ändern des Testcodes und Verwenden des zuvor definierten Dekorators können Sie eine temporäre Datenbank zum Testen zum Zeitpunkt der Testausführung auf Funktionsbasis erstellen und mit dieser Datenbank testen. Da jeder eine unabhängige Datenbank verwendet, hat dies keine Auswirkungen auf andere Testfälle.
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
Ich habe zusammengefasst, wie eine Test-DB mit FastAPI erstellt und Unittest of API mit pytest ausgeführt wird. Das Neuerstellen der Datenbank für jeden Testfall scheint die Verarbeitungsgeschwindigkeit zu verlangsamen, daher habe ich versucht, die Rollback-Methode zu ändern, aber aufgegeben. Ich denke, dass diese Methode auch dann effektiv ist, wenn es nur wenige Testfälle gibt oder wenn die Datenmenge beim Testen nicht groß ist. Ich hoffe, dieser Artikel ist hilfreich für Sie!
Refs
Recommended Posts