Create a CRUD app for your user account. Since the introduction of the test method is the main, only the following simple functions will be implemented.
--Create: User registration
In addition to FastAPI, you need pip install from the following packages.
Implement the required ones with the following directory structure.
users
├── __init__.py
├── crud.py #Function definition for query issuance
├── conftest.py #pytest fixture definition
├── database.py #Database settings
├── main.py #API definition
├── models.py #table definition
├── schemas.py #API I/O definition
└── tests
├── __init__.py
└── test_user.py #API testing
Specify the database to connect with the Database URL. Basically, you should declare it as an environment variable, but for the sake of simplicity, write it solid. The URL notation for major Databases is summarized here [https://docs.sqlalchemy.org/en/13/core/engines.html).
False}Since the part of is the setting for sqlite3, please delete it when using other Database.
```sessionlocal variable```Is in```sessionmaker```instance is,
Call will create a session instance. This is used to manage the connection with the DB. We also use session to issue SQL queries.
#### **`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()
Table definition is performed by inheriting Base defined when database is set. By defining in this way, you can easily create a table and use the ORM mapper 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 uses session to issue SQL queries. This process is prone to problems, so cut it out to make it easier to unit test. As much as possible, receive the session without inserting logic and only issue the query.
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
Defines API I / O. Here, enter email and password-> we will return the id, email, and active user of the created user. Just decide the schema and it will do Serialize and Deserialize without permission.
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
Define the CRUD API. One thing to be aware of is how to pass the session. If you declare a function or class as `` `depends``` in the argument, the result of calling it (return for function, instance for class) is passed to the argument. Using this, a session is created using SessionLocal for each request, and a connection with the database is secured. Then, the query is issued using that 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
#table creation
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
# Dependency
def get_db():
try:
db = SessionLocal() #Generate 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)
There is also an implementation that uses middleware as a method of passing a session, but since all APIs create a connection with the DB, there are harmful effects such as waste if there are many APIs that do not use the DB. Seems to be deprecated. (Reference)
Will begin the main subject.
In FastAPI, you can simply test the API with starlette.testclient.TestClient as follows.
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
Now, run pytest for automated testing.
$ pytest
However, the API connects to the production DB, so when I run the test, I add user. If you run the test twice, the email with the same name is already registered the second time, so the user creation fails and the test does not pass.
Therefore, create a database temporarily at the time of test execution so that the database for production is not affected, and it is possible to prepare a clean database every time the test is executed. In addition, we will recreate the Database for each function so that it can be used for general purposes so that it does not affect each other for each test case.
The processing required to test with a clean DB is as follows.
--Temporarily create Database for each function --Pass the Sessionmaker instance that can create a session with the Database to the test function --Delete Database at the end of each test case
If you can do these things, you can use a clean Database for each test case, leaving no trace at the end. Since this process is required regardless of the test case, we will define a fixture to perform such process in `` `conftest.py```. The implementation looks like this:
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)
By declaring SessionLocal as an argument thanks to the fixture, a clean Database is now created when the function is executed. After that, it is necessary to forcibly change the DB to which the API connects to the one for testing. I just want to complete the test code to reduce the impact.
In FastAPI, the `FastAPI.Depends``` declared in the API argument can be forcibly overwritten with ```app.dependency_overrides```. So you can change the connection destination by overwriting `main.get_db``` and rewriting it to use the sessionmaker instance for testing.
Therefore, define the following decorator.
test_user.py
from users.main import app, get_db
def temp_db(f):
def func(SessionLocal, *args, **kwargs):
#Sessionmaker instance to connect to test DB
# (SessionLocal)From fixture
def override_get_db():
try:
db = SessionLocal()
yield db
finally:
db.close()
#get to use SessionLocal received from fixture_Force db change
app.dependency_overrides[get_db] = override_get_db
# Run tests
f(*args, **kwargs)
# get_Undo db
app.dependency_overrides[get_db] = get_db
return func
Simply modify the test code and use the decorator you defined earlier to create a temporary database for testing at test execution on a function-by-function basis, and you can use that database for testing. In addition, since each database will be used independently, it will not affect other test cases.
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
I have summarized how to create a test DB with FastAPI and perform API Unittest with pytest. Recreating the DB for each test case seems to slow down the processing speed, so I tried and errored the rollback method, but I gave up. I think that this method is also effective when there are few test cases or the amount of data when testing is not large, so I hope this article will be helpful for you!
Refs
Recommended Posts