Pytest API Testing with FastAPI, SQLAlchemy, Postgres - 1/2

Posted on 2024-07-09 Edited on 2024-11-13 In Advanced , API Testing Word count in article: 4.9k Reading time ≈ 18 mins.

Building APIs is easy. Testing them is harder.

API testing involves multiple components and strays into the realm of integration testing.

APIs form an interface between the real world consumers of data and a data store.

So how do you test this interaction of components?

How do you test that your API correctly handles the incoming payload and return values?

What about making sure your API returns correct HTTP response codes?

Should you use a real database or mock the queries? Does that create a strong coupling with the source code?

What about an in-memory database like SQLite or TinyDB?

In this Pytest API Testing Masterclass, we’ll answer all your API testing questions.

We’ll go deep into building our own Mortgage Calculator API with FastAPI, SQLAlchemy, Postgres, and Pytest.

We’ll test our API using recommended Pytest functionality like fixtures, setup, teardown, and hooks.

You’ll learn to maintain a clean database state between tests and use Pytest hooks to manage sessions and fixtures to handle common objects across tests.

There’s a lot packed into this article so I encourage you to get your IDE out and follow along.

To avoid this article being insanely long I’ve split it up into 2 parts, to be published sequentially.

Hope you’re excited, let’s begin.

Why is API Testing Important?

So why is API testing important?

Isn’t it enough to build the API and hope it works? Maybe run a few checks on Postman?

That’s OK for your pet projects, but not when building enterprise-level software.

The cost of failure is too high, especially when dealing with user data.

Let’s understand why API testing is important and should be well handled during the development process.

Ensures Reliability: API testing helps verify that the API functions as expected.
Detects Issues Early: API testing can identify and fix issues before they become more significant problems, saving time and resources.
Validates Data Integrity: API testing ensures that the data exchanged between systems is accurate and consistent, maintaining data integrity across different services.
Enhances Security: API testing helps identify potential security vulnerabilities, such as unauthorized access or data breaches, and address them proactively.
Improves Performance: API testing ensures the API can handle the expected load.
Facilitates Automation: API testing increases confidence in CI/CD pipelines and automated release processes.
Reduces Manual Testing: Automated API tests reduce the need for extensive manual testing, freeing up resources and allowing testers to focus on more complex test scenarios.
Ensures Compliance: API testing helps ensure that the API complies with industry standards and regulations, which is crucial for industries like finance and healthcare.
Improves Documentation: Creating and maintaining API tests often leads to better-documented APIs, making it easier for you and your team to understand and use them correctly.
Regression and Smoke Tests: Automated API tests can ensure the API still works as expected when new changes are introduced, reducing regression. It also ensures critical paths (e.g. User Registration, Checkout) are checked and working at all times.

Most importantly, robust API testing allows you to sleep peacefully at night.

Now that you’re on-board and understand the need for robust API testing, let’s dive into the problem statement.

Problem Statement

Let’s examine the problem statement or the Statement of Work.

Build a Mortgage Calculator that calculates the monthly payment for an interest-only and repayment-type mortgage on a property.

As with anything in coding, you must get a clear problem statement to know exactly what you’re solving.

We’ll make a few assumptions here for simplicity but I encourage you to clarify every assumption in the real world.

If you’re not familiar with these types of mortgages this article covers the differences.

In short, the calculation for each is different.

In this example, we’ll build CRUD routes (Create, Read, Update, and Delete) for Properties and Mortgages. We’ll also have a custom calculations to calculate the monthly mortgage payment in the case of each - interest only and repayment mortgages.

As you can see in the ORM diagram below, mortgages are linked to a property via property_id and can be of interest_only or repayment type.

Project Setup

Let’s get started with the coding.

Clone Repo

Clone the repository containing the example code or if you prefer to start from scratch you can do so by following the similar directory structure.

.
├── .gitignore
├── README.md
└── mortgage_calculator
    ├── app
    │   ├── __init__.py
    │   ├── crud
    │   │   ├── mortgage_crud.py
    │   │   └── property_crud.py
    │   ├── custom
    │   │   ├── calculations.py
    │   │   └── db_queries.py
    │   ├── database.py
    │   ├── main.py
    │   ├── models.py
    │   ├── openapi.json
    │   ├── routes.py
    │   └── schemas.py
    ├── docker
    │   └── docker-compose.yml
    ├── poetry.lock
    ├── pyproject.toml
    └── tests
        ├── __init__.py
        ├── conftest.py
        ├── crud
        │   ├── test_crud_errors.py
        │   ├── test_mortgage_crud.py
        │   └── test_property_crud.py
        └── custom
            └── test_mortgage_payments.py

We’ll be using Poetry to manage our dependencies and virtual environment so if you’re not familiar with Poetry I encourage you to start using it.

It’s a great tool that picks the best from pip, venv, conda, and others.

Install pyenv

I use pyenv to manage my Python versions but you can use any other tool.

Let’s set the Python version in the terminal to 3.12

1 2	$ pyenv install 3.12 $ pyenv local 3.12

Install Dependencies

Our pyproject.toml file contains the dependencies.

[tool.poetry]
name = "mortgage-calculator"
version = "0.1.0"
description = "Mortgage Calculator API"
authors = []
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.12"
fastapi = "0.111.0"
uvicorn = "^0.30.1"
pydantic = "^2.7.4"
httpx = "^0.27.0"
sqlalchemy-utils = "^0.41.2"
psycopg2 = "^2.9.9"
python-dotenv = "^1.0.1"


[tool.poetry.group.dev.dependencies]
pytest = "^8.2.2"
pytest-randomly = "^3.15.0"
ruff = "^0.5.0"
black = "^24.4.2"
isort = "^5.13.2"
pytest-cov = "^5.0.0"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

To install dependencies, go ./mortgage_calculator/ and run,

1
2
3

$ poetry env use 3.12.3  
$ poetry env info # Verify   
$ poetry install

This will create a virtual environment in your directory and install the packages.

You can also do this interactively using poetry init .

Now that you have a working environment, go through the source code.

Source Code

First, let’s go through the code to understand it and then fire it up.

DB and DB Models

The SQL database models used are very simple and consist of only 2 tables — mortgages and properties.

We’ll store property details in the prpoperties table and associated mortgages in the mortgages table, joined by property_id .

Pytest API Testing ORM Diagram

The below code contains the SQLAlchemy ORM models.

mortgage_calculator/app/models.py

import uuid  
from enum import Enum  
  
from sqlalchemy import TIMESTAMP, Column  
from sqlalchemy import Enum as SQLAlchemyEnum  
from sqlalchemy import ForeignKey, Numeric, String  
from sqlalchemy.orm import relationship  
from sqlalchemy.sql import func  
from sqlalchemy_utils import UUIDType  
  
from app.database import Base  
  
  
class MortgageType(str, Enum):  
    interest_only = "interest_only"  
    repayment = "repayment"  
  
  
class PropertyOrm(Base):  
    __tablename__ = "properties"  
  
    id = Column(UUIDType(binary=False), primary_key=True, default=uuid.uuid4)  
    purchase_price = Column(Numeric(10, 2), nullable=False)  
    rental_income = Column(Numeric(10, 2), nullable=False)  
    renovation_cost = Column(Numeric(10, 2), nullable=False)  
    property_name = Column(String(255), nullable=False)  
    admin_costs = Column(Numeric(10, 2), nullable=False)  
    management_fees = Column(Numeric(10, 2), nullable=False)  
  
    createdAt = Column(  
        TIMESTAMP(timezone=True), nullable=False, server_default=func.now()  
    )  
    updatedAt = Column(TIMESTAMP(timezone=True), default=None, onupdate=func.now())  
    mortgages = relationship("MortgageOrm", back_populates="property")  
  
  
class MortgageOrm(Base):  
    __tablename__ = "mortgages"  
  
    id = Column(UUIDType(binary=False), primary_key=True, default=uuid.uuid4)  
    property_id = Column(UUIDType(binary=False), ForeignKey("properties.id"))  
    loan_to_value = Column(Numeric(5, 2), nullable=False)  
    interest_rate = Column(Numeric(5, 2), nullable=False)  
    mortgage_type = Column(SQLAlchemyEnum(MortgageType), nullable=False)  
    loan_term = Column(Numeric(5, 2), nullable=True)  
    mortgage_amount = Column(Numeric(10, 2), nullable=False)  
  
    property = relationship("PropertyOrm", back_populates="mortgages")  
  
    createdAt = Column(  
        TIMESTAMP(timezone=True), nullable=False, server_default=func.now()  
    )  
    updatedAt = Column(TIMESTAMP(timezone=True), default=None, onupdate=func.now())

Both tables have createdAt and updatedAt timestamps that auto-populate using SQLAlchemy in-built methods.

We’ve also included an Enum MortgageType to support interest_only and repayment type mortgages.

Here’s our database file. We’ll be using Postgres as our production database.

mortgage_calculator/app/database.py

import os  
from dotenv import load_dotenv  
from sqlalchemy import create_engine  
from sqlalchemy.orm import declarative_base, sessionmaker  
  
load_dotenv()  
  
# Get Postgres Environment Variables  
POSTGRES_HOST = os.getenv("POSTGRES_HOST")  
POSTGRES_USER = os.getenv("POSTGRES_USER")  
POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD")  
POSTGRES_DB = os.getenv("POSTGRES_DB")  
  
SQLALCHEMY_DATABASE_URL = (  
    f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}/{POSTGRES_DB}"  
)  
  
engine = create_engine(  
    SQLALCHEMY_DATABASE_URL,  
)  
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)  
  
Base = declarative_base()  
  
  
def get_db():  
    db = SessionLocal()  
    try:  
        yield db  
    finally:  
        db.close()
````
We’ve used a `.env` file but you can use [direnv](https://direnv.net/) or define these in the terminal if you prefer.
```env
POSTGRES_HOST="localhost"  
POSTGRES_USER="myuser"  
POSTGRES_PASSWORD="mypassword"  
POSTGRES_DB="mydatabase"

Now let’s define our Pydantic Models which represent important entities like our Property Base Model, Mortgage Base Models and their CRUD Models.

Pydantic Models and Schemas

`mortgage_calculator/app/schemas.py`

from datetime import datetime  
from enum import Enum  
from typing import List, Optional  
from uuid import UUID  
  
from pydantic import BaseModel, ConfigDict, Field  
  
from app.models import MortgageType  
  
  
class Status(Enum):  
    Success = "Success"  
    Error = "Error"  
  
  
class PropertyBaseModel(BaseModel):  
    purchase_price: Optional[float] = Field(  
        default=None,  
        json_schema_extra={  
            "example": 300000.00,  
            "description": "The purchase price of the property",  
        },  
    )  
  
    rental_income: Optional[float] = Field(  
        default=None,  
        json_schema_extra={  
            "example": 1500.00,  
            "description": "Monthly rental income from the property",  
        },  
    )  
    renovation_cost: Optional[float] = Field(  
        default=None,  
        json_schema_extra={  
            "example": 20000.00,  
            "description": "Cost of renovation for the property",  
        },  
    )  
    property_name: Optional[str] = Field(  
        default=None,  
        json_schema_extra={  
            "example": "Property 1",  
            "description": "Name of the property",  
        },  
    )  
    admin_costs: Optional[float] = Field(  
        default=None,  
        json_schema_extra={  
            "example": 500.00,  
            "description": "Administrative costs associated with buying the property",  
        },  
    )  
    management_fees: Optional[float] = Field(  
        default=None,  
        json_schema_extra={  
            "example": 100.00,  
            "description": "Monthly management fees for the property",  
        },  
    )  
    model_config = ConfigDict(from_attributes=True)  
  
  
class PropertyCreateModel(PropertyBaseModel):  
    purchase_price: float  
    rental_income: float  
    renovation_cost: float  
    property_name: str  
    admin_costs: float  
    management_fees: float  
  
  
class PropertyUpdateModel(PropertyBaseModel):  
    pass  
  
  
class PropertyModel(PropertyBaseModel):  
    id: UUID  
    createdAt: datetime  
    updatedAt: Optional[datetime]  
  
    model_config = ConfigDict(from_attributes=True)  
  
  
class PropertyResponseModel(BaseModel):  
    status: Status = Status.Success  
    message: str  
    data: PropertyModel  
  
  
class PropertyListResponseModel(BaseModel):  
    status: Status = Status.Success  
    message: str  
    data: List[PropertyModel]  
  
  
class PropertyDeleteModel(BaseModel):  
    id: UUID  
    status: Status = Status.Success  
    message: str  
  
  
# Mortgage Schemas  
class MortgageBaseModel(BaseModel):  
    loan_to_value: Optional[float] = Field(  
        default=None,  
        json_schema_extra={  
            "example": 75.0,  
            "description": "Loan to value ratio for the mortgage",  
        },  
    )  
    interest_rate: Optional[float] = Field(  
        default=None,  
        json_schema_extra={  
            "example": 2.5,  
            "description": "Interest rate for the mortgage",  
        },  
    )  
    mortgage_type: Optional[MortgageType] = Field(  
        default=None,  
        json_schema_extra={  
            "example": "Repayment",  
            "description": "Type of mortgage",  
        },  
    )  
    loan_term: Optional[int] = Field(  
        default=None,  
        json_schema_extra={  
            "example": 30,  
            "description": "Loan term in years",  
        },  
    )  
    mortgage_amount: Optional[float] = Field(  
        default=None,  
        json_schema_extra={  
            "example": 225000.00,  
            "description": "Amount of the mortgage. Calculated based on loan to value and purchase price.",  
        },  
    )  
  
    model_config = ConfigDict(from_attributes=True)  
  
  
class MortgageCreateModel(MortgageBaseModel):  
    loan_to_value: float  
    interest_rate: float  
    loan_term: int  
    mortgage_type: MortgageType  
    property_id: UUID  
  
  
class MortgageUpdateModel(MortgageBaseModel):  
    pass  
  
  
class MortgageModel(MortgageBaseModel):  
    id: UUID  
    createdAt: datetime  
    updatedAt: Optional[datetime]  
    property_id: UUID  
  
    model_config = ConfigDict(from_attributes=True)  
  
  
class MortgageResponseModel(BaseModel):  
    status: Status = Status.Success  
    message: str  
    data: MortgageModel  
  
  
class MortgageDeleteModel(BaseModel):  
    id: UUID  
    status: Status = Status.Success  
    message: str  
  
  
class MortgageListResponseModel(BaseModel):  
    status: Status = Status.Success  
    message: str  
    data: List[MortgageModel]

In the above code snippet, we have

Status Enum
PropertyBaseModel — which contains the required payload fields, all optional
PropertyCreateModel- which inherits the BaseModel and requires all the fields.
PropertyUpdateModel
PropertyDeleteModel
ResponseModel

and similarly for Mortgages.

Why should you define models?

Our goal is to validate the incoming payload before translating it into SQLAlchemy Models that eventually write to the database.

And the same goes for responses and vice-versa.

This helps us maintain strong type validation so we’re not purely dealing with raw Python objects but rather strongly typed Pydantic Classes that handle errors and type checks beautifully.

Routing and Main File

Let’s look at the various routes we’ve defined.

Our main file looks like this.

mortgage_calculator/app/main.py

from fastapi import FastAPI  
from fastapi.middleware.cors import CORSMiddleware  
  
from app import models, routes  
from app.database import engine  
  
models.Base.metadata.create_all(bind=engine)  
  
app = FastAPI()  
  
origins = [  
    "http://localhost:3000",  
]  
  
app.add_middleware(  
    CORSMiddleware,  
    allow_origins=origins,  
    allow_credentials=True,  
    allow_methods=["*"],  
    allow_headers=["*"],  
)  
  
  
app.include_router(routes.router, tags=["MortgageCalculator"], prefix="/api/v1")  
  
  
@app.get("/api/healthchecker")  
def root():  
    return {"message": "The API is LIVE!!"}

We included a simple health check endpoint to make sure the app has started and is working.

Our base endpoint is /api/v1 so on local host it would be http://localhost:port/api/v1 .

Next is our route file.

mortgage_calculator/app/routes.py

from typing import Dict  
  
from fastapi import APIRouter, Depends, status  
from sqlalchemy.orm import Session  
  
import app.schemas as schemas  
from app.crud.mortgage_crud import (  
    create_mortgage_crud,  
    delete_mortgage_crud,  
    get_mortgage_crud,  
    get_mortgages_crud,  
    update_mortgage_crud,  
)  
from app.crud.property_crud import (  
    create_property_crud,  
    delete_property_crud,  
    get_properties_crud,  
    get_property_crud,  
    update_property_crud,  
)  
from app.custom.db_queries import get_mortgage_payment  
from app.database import get_db  
  
router = APIRouter()  
  
  
@router.post(  
    "/property",  
    status_code=status.HTTP_201_CREATED,  
    response_model=schemas.PropertyResponseModel,  
)  
def create_property(  
    property: schemas.PropertyCreateModel, db: Session = Depends(get_db)  
) -> schemas.PropertyResponseModel:  
    """  
    Create a new property.  
    """  
    return create_property_crud(payload=property, db=db)  
  
  
@router.get(  
    "/property/{property_id}",  
    status_code=status.HTTP_200_OK,  
    response_model=schemas.PropertyResponseModel,  
)  
def get_property(  
    property_id: str, db: Session = Depends(get_db)  
) -> schemas.PropertyResponseModel:  
    """  
    Get a property by ID.  
    """  
    return get_property_crud(property_id=property_id, db=db)  
  
  
@router.patch(  
    "/property/{property_id}",  
    status_code=status.HTTP_202_ACCEPTED,  
    response_model=schemas.PropertyResponseModel,  
)  
def update_property(  
    property_id: str,  
    property: schemas.PropertyUpdateModel,  
    db: Session = Depends(get_db),  
) -> schemas.PropertyResponseModel:  
    """  
    Update a property by ID.  
    """  
    return update_property_crud(property_id=property_id, payload=property, db=db)  
  
  
@router.delete(  
    "/property/{property_id}",  
    status_code=status.HTTP_202_ACCEPTED,  
    response_model=schemas.PropertyDeleteModel,  
)  
def delete_property(  
    property_id: str, db: Session = Depends(get_db)  
) -> schemas.PropertyDeleteModel:  
    """  
    Delete a property by ID.  
    """  
    return delete_property_crud(property_id=property_id, db=db)  
  
  
@router.get(  
    "/property",  
    status_code=status.HTTP_200_OK,  
    response_model=schemas.PropertyListResponseModel,  
)  
def get_properties(db: Session = Depends(get_db)) -> schemas.PropertyListResponseModel:  
    """  
    Get all properties.  
    """  
    return get_properties_crud(db=db)  
  
  
@router.post(  
    "/mortgage",  
    status_code=status.HTTP_201_CREATED,  
    response_model=schemas.MortgageResponseModel,  
)  
def create_mortgage(  
    mortgage: schemas.MortgageCreateModel, db: Session = Depends(get_db)  
) -> schemas.MortgageResponseModel:  
    """  
    Create a new mortgage.  
    """  
    return create_mortgage_crud(payload=mortgage, db=db)  
  
  
@router.get(  
    "/mortgage/{mortgage_id}",  
    status_code=status.HTTP_200_OK,  
    response_model=schemas.MortgageResponseModel,  
)  
def get_mortgage(  
    mortgage_id: str, db: Session = Depends(get_db)  
) -> schemas.MortgageResponseModel:  
    """  
    Get a mortgage by ID.  
    """  
    return get_mortgage_crud(mortgage_id=mortgage_id, db=db)  
  
  
@router.patch(  
    "/mortgage/{mortgage_id}",  
    status_code=status.HTTP_202_ACCEPTED,  
    response_model=schemas.MortgageResponseModel,  
)  
def update_mortgage(  
    mortgage_id: str,  
    mortgage: schemas.MortgageUpdateModel,  
    db: Session = Depends(get_db),  
) -> schemas.MortgageResponseModel:  
    """  
    Update a mortgage by ID.  
    """  
    return update_mortgage_crud(mortgage_id=mortgage_id, payload=mortgage, db=db)  
  
  
@router.delete(  
    "/mortgage/{mortgage_id}",  
    status_code=status.HTTP_202_ACCEPTED,  
    response_model=schemas.MortgageDeleteModel,  
)  
def delete_mortgage(  
    mortgage_id: str, db: Session = Depends(get_db)  
) -> schemas.MortgageDeleteModel:  
    """  
    Delete a mortgage by ID.  
    """  
    return delete_mortgage_crud(mortgage_id=mortgage_id, db=db)  
  
  
@router.get(  
    "/mortgage",  
    status_code=status.HTTP_200_OK,  
    response_model=schemas.MortgageListResponseModel,  
)  
def get_mortgages(db: Session = Depends(get_db)) -> schemas.MortgageListResponseModel:  
    """  
    Get all mortgages.  
    """  
    return get_mortgages_crud(db=db)  
  
  
@router.post("/mortgage/{mortgage_id}/payment", response_model=Dict[str, str | float])  
def calculate_mortgage_payment(  
    mortgage_id: str, db: Session = Depends(get_db)  
) -> Dict[str, str | float]:  
    """  
    Retrieve the monthly payment for a given mortgage.  
    """  
    return get_mortgage_payment(mortgage_id, db)

Property Routes

Create Property (POST)
Get Property (GET)
Update Property (PATCH)
Delete Property (DELETE)
Get Properties (GET)

Similarly for mortgages.

All property routes are prefixed by /property while mortgage routes are prefixed by /mortgage .

Now you’ll notice something — in most FastAPI examples, the routing logic i.e validating the payload and writing to the database happens within this file.

That’s OK for small applications but as you grow more complex your routing logic gets complicated, hence it’s a good idea to split this out into a separate file.

You’ll see why shortly.

CRUD Logic

mortgage_calculator/app/crud/property_crud.py

from fastapi import Depends, HTTPException, status  
from sqlalchemy.exc import IntegrityError, SQLAlchemyError  
from sqlalchemy.orm import Session  
  
import app.models as models  
import app.schemas as schemas  
from app.database import get_db  
  
  
def create_property_crud( payload: schemas.PropertyCreateModel, db: Session = Depends(get_db)):  
    try:  
        new_property = models.PropertyOrm(**payload.model_dump())  
        db.add(new_property)  
        db.commit()  
        db.refresh(new_property)  
  
        property_data = schemas.PropertyModel.model_validate(new_property)  
  
        return schemas.PropertyResponseModel(  
            status=schemas.Status.Success,  
            message="Property created successfully.",  
            data=property_data,  
        )  
    except IntegrityError:  
        db.rollback()  
        raise HTTPException(  
            status_code=status.HTTP_409_CONFLICT,  
            detail="A property with the given details already exists.",  
        )  
    except Exception as e:  
        db.rollback()  
        raise HTTPException(  
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,  
            detail=f"An error occurred while creating the property: {str(e)}",  
        )  
  
  
def get_property_crud(property_id: str, db: Session = Depends(get_db)):  
    property_data = (  
        db.query(models.PropertyOrm)  
        .filter(models.PropertyOrm.id == property_id)  
        .first()  
    )  
  
    if not property_data:  
        raise HTTPException(  
            status_code=status.HTTP_404_NOT_FOUND,  
            detail="Property not found.",  
        )  
  
    try:  
        return schemas.PropertyResponseModel(  
            status=schemas.Status.Success,  
            message="Property retrieved successfully.",  
            data=schemas.PropertyModel.model_validate(property_data),  
        )  
    except SQLAlchemyError as e:  
        raise HTTPException(  
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,  
            detail="An unexpected error occurred while retrieving the property.",  
        ) from e  
  
  
def update_property_crud( property_id: str, payload: schemas.PropertyUpdateModel, db: Session):  
    property_query = db.query(models.PropertyOrm).filter(  
        models.PropertyOrm.id == property_id  
    )  
    db_property = property_query.first()  
  
    if not db_property:  
        raise HTTPException(  
            status_code=status.HTTP_404_NOT_FOUND, detail="Property not found."  
        )  
  
    try:  
        # Prepare update data from the payload, only including fields that are set  
        update_data = payload.model_dump(exclude_unset=True)  
        if update_data:  
            property_query.update(update_data, synchronize_session="evaluate")  
            db.commit()  
            db.refresh(db_property)  
  
            # Convert the updated ORM model back to a Pydantic model  
            return schemas.PropertyResponseModel(  
                status=schemas.Status.Success,  
                message="Property updated successfully.",  
                data=schemas.PropertyModel.model_validate(db_property),  
            )  
        else:  
            raise HTTPException(  
                status_code=status.HTTP_400_BAD_REQUEST,  
                detail="No valid fields provided for update.",  
            )  
    except IntegrityError as e:  
        db.rollback()  
        raise HTTPException(  
            status_code=status.HTTP_409_CONFLICT,  
            detail="A property with the given details already exists.",  
        ) from e  
    except SQLAlchemyError as e:  
        db.rollback()  
        raise HTTPException(  
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,  
            detail=f"An error occurred while updating the property: {str(e)}",  
        ) from e  
  
  
def delete_property_crud(property_id: str, db: Session = Depends(get_db)):  
    try:  
        property_query = db.query(models.PropertyOrm).filter(  
            models.PropertyOrm.id == property_id  
        )  
        property_ = property_query.first()  
        if not property_:  
            raise HTTPException(  
                status_code=status.HTTP_404_NOT_FOUND,  
                detail=f"No property found with ID: {property_id}",  
            )  
        property_query.delete(synchronize_session=False)  
        db.commit()  
        return schemas.PropertyDeleteModel(  
            id=property_id,  
            status=schemas.Status.Success,  
            message="Property deleted successfully.",  
        )  
    except Exception as e:  
        db.rollback()  
        raise HTTPException(  
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,  
            detail="An unexpected error occurred while deleting the property.",  
        ) from e  
  
  
def get_properties_crud( db: Session = Depends(get_db), limit: int = 10, page: int = 1, search: str = ""):  
    skip = (page - 1) * limit  
  
    properties = db.query(models.PropertyOrm).limit(limit).offset(skip).all()  
    return schemas.PropertyListResponseModel(  
        status=schemas.Status.Success,  
        message="Properties retrieved successfully.",  
        data=[  
            schemas.PropertyModel.model_validate(property_) for property_ in properties  
        ],  
    )

Here we have the logic for Property CRUD operations.

There is a constant conversion from Pydantic object/model to SQLAlchemy ORM object and vice versa.

As mentioned above this ensures we’re always working with strongly type-checked classes.

You’ll also notice we’re handling errors like

404 — Object not found
409 — Conflict
500 — Internal Server Error (which is not great tbh)
200 — Successful response
201 — Created successfully

You can find the complete list of HTTP response codes here. In building enterprise-level APIs it’s very important to take this seriously and make sure it’s tested thoroughly.

Let’s look at how we’ve written the CRUD operations for mortgages.

mortgage_calculator/app/crud/mortgage_crud.py

from fastapi import Depends, HTTPException, status  
from sqlalchemy.exc import IntegrityError, SQLAlchemyError  
from sqlalchemy.orm import Session  
  
import app.models as models  
import app.schemas as schemas  
from app.database import get_db  
  
  
def create_mortgage_crud( payload: schemas.MortgageCreateModel, db: Session = Depends(get_db)):  
    try:  
        # Create a new mortgage object from the payload  
        new_mortgage = models.MortgageOrm(**payload.model_dump())  
  
        # Check whether the property exists in the DB  
        property_data = (  
            db.query(models.PropertyOrm)  
            .filter(models.PropertyOrm.id == payload.property_id)  
            .first()  
        )  
        if not property_data:  
            raise HTTPException(  
                status_code=status.HTTP_404_NOT_FOUND,  
                detail="Property not found.",  
            )  
  
        # Get Purchase Price of the property and calculate mortgage amount (LTV * Purchase Price)  
        purchase_price = float(property_data.purchase_price)  # Convert to float  
        loan_to_value = float(new_mortgage.loan_to_value)  # Ensure this is float  
        mortgage_amount = (purchase_price * loan_to_value) / 100  
        new_mortgage.mortgage_amount = mortgage_amount  
  
        db.add(new_mortgage)  
        db.commit()  
        db.refresh(new_mortgage)  
  
        # Validate the new mortgage object  
        mortgage_data = schemas.MortgageModel.model_validate(new_mortgage)  
  
        return schemas.MortgageResponseModel(  
            status=schemas.Status.Success,  
            message="Mortgage created successfully.",  
            data=mortgage_data,  
        )  
    except IntegrityError:  
        db.rollback()  
        raise HTTPException(  
            status_code=status.HTTP_409_CONFLICT,  
            detail="A mortgage with the given details already exists.",  
        )  
  
  
def get_mortgage_crud(mortgage_id: str, db: Session = Depends(get_db)):  
    mortgage_data = (  
        db.query(models.MortgageOrm)  
        .filter(models.MortgageOrm.id == mortgage_id)  
        .first()  
    )  
  
    if not mortgage_data:  
        raise HTTPException(  
            status_code=status.HTTP_404_NOT_FOUND,  
            detail="Mortgage not found.",  
        )  
  
    try:  
        return schemas.MortgageResponseModel(  
            status=schemas.Status.Success,  
            message="Mortgage retrieved successfully.",  
            data=schemas.MortgageModel.model_validate(mortgage_data),  
        )  
    except SQLAlchemyError as e:  
        raise HTTPException(  
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,  
            detail="An unexpected error occurred while retrieving the mortgage.",  
        ) from e  
  
  
def update_mortgage_crud( mortgage_id: str, payload: schemas.MortgageUpdateModel, db: Session):  
    mortgage_query = db.query(models.MortgageOrm).filter(  
        models.MortgageOrm.id == mortgage_id  
    )  
    db_mortgage = mortgage_query.first()  
  
    if not db_mortgage:  
        raise HTTPException(  
            status_code=status.HTTP_404_NOT_FOUND, detail="Mortgage not found."  
        )  
  
    try:  
        # Prepare update data from the payload, only including fields that are set  
        update_data = payload.model_dump(exclude_unset=True)  
        if update_data:  
            mortgage_query.update(update_data, synchronize_session="evaluate")  
            db.commit()  
            db.refresh(db_mortgage)  
  
            # Convert the updated ORM model back to a Pydantic model  
            return schemas.MortgageResponseModel(  
                status=schemas.Status.Success,  
                message="Mortgage updated successfully.",  
                data=schemas.MortgageModel.model_validate(db_mortgage),  
            )  
        else:  
            raise HTTPException(  
                status_code=status.HTTP_400_BAD_REQUEST,  
                detail="No valid fields provided for update.",  
            )  
    except IntegrityError as e:  
        db.rollback()  
        raise HTTPException(  
            status_code=status.HTTP_409_CONFLICT,  
            detail="A mortgage with the given details already exists.",  
        ) from e  
    except SQLAlchemyError as e:  
        db.rollback()  
        raise HTTPException(  
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,  
            detail=f"An error occurred while updating the mortgage: {str(e)}",  
        ) from e  
  
  
def delete_mortgage_crud(mortgage_id: str, db: Session = Depends(get_db)):  
    try:  
        mortgage_query = db.query(models.MortgageOrm).filter(  
            models.MortgageOrm.id == mortgage_id  
        )  
        mortgage_ = mortgage_query.first()  
        if not mortgage_:  
            raise HTTPException(  
                status_code=status.HTTP_404_NOT_FOUND,  
                detail=f"No mortgage found with ID: {mortgage_id}",  
            )  
        mortgage_query.delete(synchronize_session=False)  
        db.commit()  
        return schemas.MortgageDeleteModel(  
            id=mortgage_id,  
            status=schemas.Status.Success,  
            message="Mortgage deleted successfully.",  
        )  
    except Exception as e:  
        db.rollback()  
        raise HTTPException(  
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,  
            detail="An unexpected error occurred while deleting the mortgage.",  
        ) from e  
  
  
def get_mortgages_crud( db: Session = Depends(get_db), limit: int = 10, page: int = 1, search: str = ""):  
    skip = (page - 1) * limit  
  
    mortgages = db.query(models.MortgageOrm).limit(limit).offset(skip).all()  
    return schemas.MortgageListResponseModel(  
        status=schemas.Status.Success,  
        message="Mortgages retrieved successfully.",  
        data=[schemas.MortgageModel.model_validate(mortgage) for mortgage in mortgages],  
    )

If you observe the first method — Create Mortgage.

We have to make sure the property exists in the database before creating/associating a mortgage with it. So we perform simple checks.

# Check whether the property exists in the DB  
        property_data = (  
            db.query(models.PropertyOrm)  
            .filter(models.PropertyOrm.id == payload.property_id)  
            .first()  
        )

Checks like this can get quite complex so it’s cleaner to separate business logic from FastAPI boilerplate code.

Custom Business Logic

We also have to perform custom calculations to get the monthly mortgage payments for the 2 types of mortgages — repayment and interest_only .

mortgage_calculator/app/custom/calculations.py

def calculate_interest_only_payment( loan_amount: float, annual_interest_rate: float) -> float:  
    """  
    Calculate the monthly interest payment for an interest-only mortgage.  
  
    :param loan_amount: The total loan amount (principal).  
    :param annual_interest_rate: The annual interest rate as a percentage.  
    :return: Monthly interest payment.  
    """  
    monthly_interest_rate = annual_interest_rate / 100 / 12  
    return round(loan_amount * monthly_interest_rate, 2)  
  
  
def calculate_repayment_mortgage_payment( loan_amount: float, annual_interest_rate: float, loan_term_years: int) -> float:  
    """  
    Calculate the monthly payment for a repayment mortgage.  
  
    :param loan_amount: The total loan amount (principal).  
    :param annual_interest_rate: The annual interest rate as a percentage.  
    :param loan_term_years: The term of the loan in years.  
    :return: Monthly payment for the full term of the loan.  
    """  
    monthly_interest_rate = annual_interest_rate / 100 / 12  
    total_payments = loan_term_years * 12  
    if monthly_interest_rate == 0:  # This avoids division by zero if interest rate is 0  
        return loan_amount / total_payments  
    monthly_payment = (  
        loan_amount  
        * (monthly_interest_rate * (1 + monthly_interest_rate) ** total_payments)  
        / ((1 + monthly_interest_rate) ** total_payments - 1)  
    )  
    return round(monthly_payment, 2)

Here you can see we have a couple of formulas to perform the calculation.

Let’s see how to use these in the API itself.

mortgage_calculator/app/custom/db_queries.py

from fastapi import HTTPException, status  
from sqlalchemy.orm import Session  
  
from app.custom.calculations import (calculate_interest_only_payment,  
                                     calculate_repayment_mortgage_payment)  
from app.models import MortgageOrm, PropertyOrm  
from app.schemas import MortgageType  
  
  
def get_mortgage_payment(mortgage_id: str, db: Session):  
    mortgage = db.query(MortgageOrm).filter(MortgageOrm.id == mortgage_id).first()  
    if not mortgage:  
        raise HTTPException(  
            status_code=status.HTTP_404_NOT_FOUND, detail="Mortgage not found."  
        )  
  
    property = (  
        db.query(PropertyOrm).filter(PropertyOrm.id == mortgage.property_id).first()  
    )  
    if not property:  
        raise HTTPException(  
            status_code=status.HTTP_404_NOT_FOUND,  
            detail="Associated property not found.",  
        )  
  
    if mortgage.mortgage_type == MortgageType.interest_only.value:  
        monthly_payment = calculate_interest_only_payment(  
            mortgage.mortgage_amount, mortgage.interest_rate  
        )  
    elif mortgage.mortgage_type == MortgageType.repayment.value:  
        # Assuming a fixed loan term, e.g., 30 years. This could also be dynamically fetched or adjusted.  
        monthly_payment = calculate_repayment_mortgage_payment(  
            mortgage.mortgage_amount, mortgage.interest_rate, mortgage.loan_term  
        )  
    else:  
        raise HTTPException(  
            status_code=status.HTTP_400_BAD_REQUEST, detail="Unsupported mortgage type."  
        )  
  
    return {"mortgage_id": str(mortgage.id), "monthly_payment": float(monthly_payment)}

We have a get_mortgage_payment method that looks up a mortgage_id field and checks that it exists in the database and then gets the associated property.

In the case of interest-only, we pass the relevant mortgage amount and interest.

We do something similar for repayment however we also need the repayment term.

If you go back to routes.py you’ll see the endpoint.

mortgage_calculator/app/routes.py

@router.post("/mortgage/{mortgage_id}/payment", response_model=Dict[str, str | float])  
def calculate_mortgage_payment( mortgage_id: str, db: Session = Depends(get_db)) -> Dict[str, str | float]:  
    """  
    Retrieve the monthly payment for a given mortgage.  
    """  
    return get_mortgage_payment(mortgage_id, db)

This means we can fetch the monthly payment for any mortgage that’s created in the database.

Now that we’ve gone through the source code, it’s time to talk about the database. As mentioned before, we’ll be using Postgres (run locally on Docker).

Start Postgres Database with Docker

I’ve included a Docker Compose File to help you get started with setting up a simple Postgres container locally.

If you don’t have experience with Docker I suggest reading up on that or watching a few YouTube videos.

mortgage_calculator/docker/docker-compose.yml

services:  
  db:  
    image: postgres:latest  
    container_name: postgres_local  
    environment:  
      POSTGRES_USER: myuser  
      POSTGRES_PASSWORD: mypassword  
      POSTGRES_DB: mydatabase  
    ports:  
      - "5432:5432"  
    restart: always  
    volumes:  
      - postgres_data:/var/lib/postgresql/data  
volumes:  
  postgres_data:

Of course, feel free to use a custom username and password.

Start the Postgres Container

1	$ docker-compose up -d

Pytest API Testing Docker Compose Up

This is going to be our production database, however, in real applications, you’d want to use Google Cloud SQL, Azure SQL, AWS RDS, or other fully managed databases rather than managing your own.

Set Environment Variables

You can set the database URL environment variable manually in the terminal, use a .env file or direnv — whatever your preferred way.

In this example let’s use a .env file.

mortgage_calculator/app/.env

POSTGRES_HOST=localhost
POSTGRES_USER=myuser  
POSTGRES_PASSWORD=mypassword 
POSTGRES_DB=mydatabase

Now our app with take the values during runtime.

Start API

We haven’t discussed tests at all. You may find this strange as this website is all about testing, however this is not a TDD article.

Ideally, you want to develop and test at the same time (to me the order doesn’t matter much) but you don’t want your tests to be lagging behind.

It’s often a good idea to write tests as you develop each endpoint to make sure it works and you haven’t broken an existing endpoint when introducing a new one.

Let’s fire up the API just to make sure there are no errors and the SQLAlchemy initialization is successful.

Start the app with the following command.

1	$ poetry run uvicorn app.main:app --host localhost --port 8000 --reload

Assuming no start-up errors.

Pytest API Testing Uvicorn Start

Swagger

Let’s navigate to our health checker endpoint — http://localhost:8000/api/healthchecker

Pytest API Testing Health Checker

This works.

Now let’s quickly check Postgres to see if our tables are created.

You can use any SQL Client, I’ve used DBeaver.

Pytest API Testing DBeaver

Pytest API Testing DBeaver Tables

Our 2 tables were successfully created.

Let’s quickly check out the Swagger at http://localhost:8000/docs

Pytest API Testing Swagger

Pytest API Testing Swagger 2

Pytest API Testing Swagger 3

FastAPI provides a beautiful Swagger with a complete payload and schema models including examples of successful responses and errors. Super neat!

Pytest API Testing Swagger 4

Pytest API Testing Swagger 5

You could use the Swagger to perform initial tests however I prefer to use Postman given we can save the collection, use variables, etc.

Postman Collection

This article covers how to export your Swagger into Postman. You want to do this automatically instead of creating each endpoint manually.

You can even paste your raw Swagger file into the import option on Postman.

In Postman, navigate to your collection and set the baseUrl to localhost:8000.

You can now easily test the API endpoints.

HealthChecker

Pytest API Testing Postman Health Checker

Create Property

It’s cool that the payload template is already available.

Pytest API Testing Postman Create Property

Get Property

Copy the id and add it to the GET request

Pytest API Testing Postman Get Property

Update Property

Let’s update the renovation cost and management fees.

Pytest API Testing Postman Update Property

We successfully updated the 2 fields. Let’s do a GET property request just to make sure.

Pytest API Testing Postman Get Property 2

Yep, all good. Note the change to the updatedAt timestamp.

You can also set the property_id field as a variable within Postman.

Create Mortgage

Pytest API Testing Postman Create Mortgage

We can nicely use the property_id field within the payload.

Get Mortgage

Let’s get the mortgage to make sure it’s created correctly and references the property.

Pytest API Testing Postman Get Mortgage

Note that I’ve set the mortgage_id as a Postman variable too.

You can see we’ve also calculated and stored the mortgage_amount in the backend which is 75% of 500,000 (375,000).

Calculate Mortgage Payment

How about we calculate the mortgage payment for this?

Pytest API Testing Postman Calculate Mortgage Payment

This can be quickly verified by

1
2
3

Monthly Payment = (Interest Rate * Mortgage Amount / 12)

Monthly Payment = (0.046 * 375,000) / 12 = 1437.50.

Which gives us confidence in our calculation.

Update Mortgage Interest Rate

Let’s say inflation has risen and so the Fed has increased rates to curb borrowing. It’s now risen from 4.6% to 6%.

We update the mortgage.

Pytest API Testing Postman Update Mortgage

Calculate New Monthly Payment

Pytest API Testing Postman Calculate New Monthly Payment

Our monthly payment has now gone up to 1875.0 (quite a steep increase).

Delete Mortgage and Property

Pytest API Testing Postman Delete Mortgage

Pytest API Testing Postman Delete Property

You can of course play around and delete a mortgage, add a new repayment type mortgage, and calculate the monthly payment.

We’ll cover this in the next article when writing tests.

Conclusion

OK this is it for article 1 of this 2 post series.

In this article, you learned the need for robust API testing and it’s importance.

Next we explored a fairly simple yet powerful Mortgage Calculator that allows us to create properties, associate a mortgage to that property and lastly, calculate the monthly payment.

You learned how to define SQLAlchemy ORM models for the 2 tables as well as Pydantic Models to handle objects and entities like Property and Mortgage.

Lastly, you learned how to define API routes and the importance of separating business logic from boilerplate API code.

We also learned how to export the OpenAPI JSON schema from the Swagger and import it into Postman to run our manual checks.

In the next article of this series, we’ll go deep into writing unit and integration tests to validate various workflows.

We’ll discuss whether you should mock the database, use an in-memory database like SQLite or use a full-fledged Postgres database.

Stay tuned for the next one.

If you have any ideas for improvement or like me to cover any topics please message me on Twitter, GitHub, or Email.

Till the next time… Cheers!

Additional Reading

Example Code Used in this Article: GitHub
How To Run Pytest With Poetry (A Step-by-Step Guide)
Maximizing Quality - A Comprehensive Look at Testing in Software Development
Building And Testing FastAPI CRUD APIs With Pytest (Hands-On Tutorial)
How To Test Database Transactions With Pytest And SQLModel
Python Unit Testing Best Practices For Building Reliable Applications
Comprehensive Step-by-Step Guide to Testing Django REST APIs with Pytest
13 Proven Ways To Improve Test Runtime With Pytest