Create a clean DB for testing with FastAPI and unittest the API with pytest

things to do

  1. [Implementing a simple CRUD application using sqlite3 with FastAPI](#Implementing a simple CRUD application using sqlite3 with FastAPI)
  2. [Create a DB for each test case and perform an API unit test with pytest without affecting other test cases or production DB](#Create a DB for each test case and other test cases And unittest the API with pytest without affecting the production DB)

1. Implement a simple CRUD application using sqlite3 with FastAPI

CRUD application

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

Preparation

In addition to FastAPI, you need pip install from the following packages.

Directory structure

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

Database settings (database.py)

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 (models.py)

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)

Function definition for query issuance (crud.py)

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

API I / O definition (schemas.py)

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

API definition (main.py)

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)

2. Create a DB for each test case and perform API unit test without affecting other test cases or production DB.

Will begin the main subject.

API testing (when using a production DB)

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.

Create and delete a clean DB for testing

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)

Change the DB to which the API connects during testing

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

in conclusion

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

Create a clean DB for testing with FastAPI and unittest the API with pytest
Create a model to store information from the Google Books API for intuitive handling and testing
Create a tweet heatmap with the Google Maps API
Create a Twitter BOT with the GoogleAppEngine SDK for Python
Create a social integration API for smartphone apps with Django
Parse the Researchmap API in Python and automatically create a Word file for the achievement list
Get the trading price of virtual currency and create a chart with API of Zaif exchange
Create a CRUD API using FastAPI
Create an alias for Route53 to CloudFront with the AWS API
Create a striped illusion with gamma correction for Python3 and openCV3
Create a color picker for the color wheel with Python + Qt (PySide)
Create a REST API to operate dynamodb with the Django REST Framework
Create and return a CP932 CSV file for Excel with Chalice
You can do it in 5 minutes !? Create a face detection API with FastAPI and OpenCV and publish it on Heroku
Create a command to search for similar compounds from the target database with RDKit and check the processing time
Create a typed web app with Python's web framework "Fast API" and TypeScript / OpenAPI-Technology stack for machine learning web apps
Create an API to convert PDF files to TIF images with FastAPI and Docker
Create a REST API using the model learned in Lobe and TensorFlow Serving.
Create a simple reception system with the Python serverless framework Chalice and Twilio
Zip-compress any file with the [shell] command to create a file and delete the original file.
Try out the touch of data-driven testing with Selenium Python Bindings and py.test
Created a Python wrapper for the Qiita API
Create a LINE BOT with Minette for Python
Create a dashboard for Network devices with Django!
Create a translation tool with the Translate Toolkit
Develop a web API that returns data stored in DB with Django and SQLite
Build API server for checking the operation of front implementation with python3 and Flask
Create a native GUI app with Py2app and Tkinter
How to create a submenu with the [Blender] plugin
Create a QR code for the URL on Linux
[Boto3] Search for Cognito users with the List Users API
Create a batch of images and inflate with ImageDataGenerator
Get comments and subscribers with the YouTube Data API
Create a Layer for AWS Lambda Python with Docker
Create an audio file with the text-to-speech function with Google Text To Speak and check the text as a guide for the speech for 3 minutes.
[Linux] Create a self-signed certificate with Docker and apache
A script that makes it easy to create rich menus with the LINE Messaging API
Create a script for your Pepper skill in a spreadsheet and load SayText directly from the script
Try hitting the Twitter API quickly and easily with Python
Create a new csv with pandas based on the local csv
A note about hitting the Facebook API with the Python SDK
Create a web surveillance camera with Raspberry Pi and OpenCV
Create a child account for connect with Stripe in Python
Probably the easiest way to create a pdf with Python3
[Python] Create a date and time list for a specified period
Create applications, register data, and share with a single email
Let's create a PRML diagram with Python, Numpy and matplotlib.
Installation procedure for Python and Ansible with a specific version
Create a real-time auto-reply bot using the Twitter Streaming API
Create a partition and then install the Raspberry Pi OS
Streamline information gathering with the Twitter API and Slack bots
Tornado-Let's create a Web API that easily returns JSON with JSON
Create a deploy script with fabric and cuisine and reuse it
Create a web API that can deliver images with Django
Make a BLE thermometer and get the temperature with Pythonista3
I want to create a Dockerfile for the time being.
[Python] Create a screen for HTTP status code 403/404/500 with Django
I tried hitting the Google API with Ruby and Python-Make the database a Spreadsheet and manage it with Google Drive
I tried to create serverless batch processing for the first time with DynamoDB and Step Functions
Understand the probabilities and statistics that can be used for progress management with a python program