Integrating the Masonite ORM with FastAPI

Posted by Oluwole Majiyagbe Oluwole Majiyagbe Last updated January 25th, 2023

In this tutorial, you'll learn how to use the Masonite ORM with FastAPI.

Contents

Objectives

By the end of this tutorial, you should be able to:

  1. Integrate the Masonite ORM with FastAPI
  2. Use the Masonite ORM to interact with Postgres, MySQL, and SQLite
  3. Declare relationships in your database application with the Masonite ORM
  4. Test a FastAPI application with pytest

Why Use the Masonite ORM

The Masonite ORM is a clean, easy-to-use, object relational mapping library built for the Masonite web framework. The Masonite ORM builds on the Orator ORM, an Active Record ORM, which is heavily inspired by Laravel's Eloquent ORM.

Masonite ORM was developed to be a replacement to Orator ORM as Orator no longer receives updates and bug fixes.

Although it's designed to be used in a Masonite web project, you can use the Masonite ORM with other Python web frameworks or projects.

FastAPI

FastAPI is a modern, high-performance, batteries-included Python web framework that's perfect for building RESTful APIs. It can handle both synchronous and asynchronous requests and has built-in support for data validation, JSON serialization, authentication and authorization, and OpenAPI.

For more on FastAPI, review our FastAPI summary page.

What We're Building

We're going to be building a simple blog application with the following models:

  1. Users
  2. Posts
  3. Comments

Users will have a one-to-many relationship with Posts while Posts will also have a one-to-many relationship with Comments.

API endpoints:

  1. /api/v1/users - get details for all users
  2. /api/v1/users/<user_id> - get a single user's details
  3. /api/v1/posts - get all posts
  4. /api/v1/posts/<post_id> - get a single post
  5. /api/v1/posts/<post_id>/comments - get all comments from a single post

Project Setup

Create a directory to hold your project called "fastapi-masonite":

$ mkdir fastapi-masonite
$ cd fastapi-masonite

Create a virtual environment and activate it:

$ python3.10 -m venv .env
$ source .env/bin/activate

(.env)$

Feel free to swap out virtualenv and Pip for Poetry or Pipenv. For more, review Modern Python Environments.

Create a requirements.txt file and add the following requirements to it:

fastapi==0.89.1
uvicorn==0.20.0

Uvicorn is an ASGI (Asynchronous Server Gateway Interface) compatible server that will be used for starting up FastAPI.

Install the requirements:

(.env)$ pip install -r requirements.txt

Create a main.py file in the root folder of your project and add the following lines:

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def say_hello():
    return {"msg": "Hello World"}

Run the FastAPI server with the following command:

(.env)$ uvicorn main:app --reload

Open your web browser of choice and navigate to http://127.0.0.1:8000. You should see the following JSON response:

{
    "msg": "Hello World"
}

Masonite ORM

Add the following requirements to the requirements.txt file:

masonite-orm==2.18.6
psycopg2-binary==2.9.5

Install the new dependencies:

(.env)$ pip install -r requirements.txt

Create the following folders:

models
databases/migrations
config

The "models" folder will contain our model files, the "databases/migrations" folder will contain our migration files, and the "config" folder will hold our Masonite Database configuration file.

Database Config

Inside the "config" folder, create a database.py file. This file is required for the Masonite ORM as this is where we declare our database configurations.

For more info, visit the docs.

Within the database.py file, we need to add the DATABASE variable plus some connection information, import the ConnectionResolver from masonite-orm.connections, and register the connection details:

# config/database.py

from masoniteorm.connections import ConnectionResolver

DATABASES = {
  "default": "postgres",
  "mysql": {
    "host": "127.0.0.1",
    "driver": "mysql",
    "database": "masonite",
    "user": "root",
    "password": "",
    "port": 3306,
    "log_queries": False,
    "options": {
      #
    }
  },
  "postgres": {
    "host": "127.0.0.1",
    "driver": "postgres",
    "database": "test",
    "user": "test",
    "password": "test",
    "port": 5432,
    "log_queries": False,
    "options": {
      #
    }
  },
  "sqlite": {
    "driver": "sqlite",
    "database": "db.sqlite3",
  }
}

DB = ConnectionResolver().set_connection_details(DATABASES)

Here, we defined three different database settings:

  1. MySQL
  2. Postgres
  3. SQLite

We set the default connection to Postgres.

Note: Make sure you have a Postgres database up and running. If you want to use MySQL, change the default connection to mysql.

Masonite Models

To create a new boilerplate Masonite model, run the following masonite-orm command from the project root folder in your terminal:

(.env)$ masonite-orm model User --directory models

You should see a success message:

Model created: models/User.py

So, this command should have created a User.py file in the "models" directory with the following content:

""" User Model """

from masoniteorm.models import Model


class User(Model):
    """User Model"""

    pass

If you receive a FileNotFoundError, check to make sure that the "models" folder exists.

Run the same commands for the Posts and Comments models:

(.env)$ masonite-orm model Post --directory models
> Model created: models/Post.py

(.env)$ masonite-orm model Comment --directory models
> Model created: models/Comment.py

Next, we can create the initial migrations:

(.env)$ masonite-orm migration migration_for_user_table --create users

We added the --create flag to tell Masonite that the migration file to be created is for our users table and the database table should be created when the migration is run.

In the "databases/migration" folder, a new file should have been created:

<timestamp>_migration_for_user_table.py

Content:

"""MigrationForUserTable Migration."""

from masoniteorm.migrations import Migration


class MigrationForUserTable(Migration):
    def up(self):
        """
        Run the migrations.
        """
        with self.schema.create("users") as table:
            table.increments("id")

            table.timestamps()

    def down(self):
        """
        Revert the migrations.
        """
        self.schema.drop("users")

Create the remaining migration files:

(.env)$ masonite-orm migration migration_for_post_table --create posts
> Migration file created: databases/migrations/2022_05_04_084820_migration_for_post_table.py

(.env)$ masonite-orm migration migration_for_comment_table --create comments
> Migration file created: databases/migrations/2022_05_04_084833_migration_for_comment_table.py

Next, let's populate the fields for each of our database tables.

Database Tables

The users table should have the following fields:

  1. Name
  2. Email (unique)
  3. Address (Optional)
  4. Phone Number (Optional)
  5. Sex (Optional)

Change the migration file associated with the Users model to:

"""MigrationForUserTable Migration."""

from masoniteorm.migrations import Migration


class MigrationForUserTable(Migration):
    def up(self):
        """
        Run the migrations.
        """
        with self.schema.create("users") as table:
            table.increments("id")
            table.string("name")
            table.string("email").unique()
            table.text("address").nullable()
            table.string("phone_number", 11).nullable()
            table.enum("sex", ["male", "female"]).nullable()
            table.timestamps()

    def down(self):
        """
        Revert the migrations.
        """
        self.schema.drop("users")

For more on the table methods and column types, review Schema & Migrations from the docs.

Next, update the fields for the Posts and Comments models, taking note of the fields.

Posts:

"""MigrationForPostTable Migration."""

from masoniteorm.migrations import Migration


class MigrationForPostTable(Migration):
    def up(self):
        """
        Run the migrations.
        """
        with self.schema.create("posts") as table:
            table.increments("id")
            table.integer("user_id").unsigned()
            table.foreign("user_id").references("id").on("users")
            table.string("title")
            table.text("body")
            table.timestamps()

    def down(self):
        """
        Revert the migrations.
        """
        self.schema.drop("posts")

Comments:

"""MigrationForCommentTable Migration."""

from masoniteorm.migrations import Migration


class MigrationForCommentTable(Migration):
    def up(self):
        """
        Run the migrations.
        """
        with self.schema.create("comments") as table:
            table.increments("id")
            table.integer("user_id").unsigned().nullable()
            table.foreign("user_id").references("id").on("users")
            table.integer("post_id").unsigned().nullable()
            table.foreign("post_id").references("id").on("posts")
            table.text("body")
            table.timestamps()

    def down(self):
        """
        Revert the migrations.
        """
        self.schema.drop("comments")

Take note of:

table.integer("user_id").unsigned()
table.foreign("user_id").references("id").on("users")

The lines above create a foreign key from the posts/comments table to the users table. The user_id column references the id column on the users table

To apply the migrations, run the following command in your terminal:

(.env)$ masonite-orm migrate

You should see success messages concerning each of your migrations:

Migrating: 2022_05_04_084807_migration_for_user_table
Migrated: 2022_05_04_084807_migration_for_user_table (0.08s)
Migrating: 2022_05_04_084820_migration_for_post_table
Migrated: 2022_05_04_084820_migration_for_post_table (0.04s)
Migrating: 2022_05_04_084833_migration_for_comment_table
Migrated: 2022_05_04_084833_migration_for_comment_table (0.02s)

Thus far, we've added and referenced the foreign keys in our table, which have been created in the database. We still need to tell Masonite what type of relationship each model has to one another, though.

Table Relationships

To define a one-to-many relationship, we need to import in has_many from masoniteorm.relationships within models/User.py and add it as decorators to our functions:

# models/User.py

from masoniteorm.models import Model
from masoniteorm.relationships import has_many


class User(Model):
    """User Model"""

    @has_many("id", "user_id")
    def posts(self):
        from .Post import Post

        return Post

    @has_many("id", "user_id")
    def comments(self):
        from .Comment import Comment

        return Comment

Do note that the has_many takes two arguments which are:

  1. The name of the primary key column on the main table which will be referenced in another table
  2. The name of the column which will serve as a reference to the foreign key

In the users table, the id is the primary key column while the user_id is the column in the posts table which references the users table record.

Do the same for models/Post.py:

# models/Post.py

from masoniteorm.models import Model
from masoniteorm.relationships import has_many


class Post(Model):
    """Post Model"""

    @has_many("id", "post_id")
    def comments(self):
        from .Comment import Comment

        return Comment

With the database configured, let's wire up our API with FastAPI.

FastAPI RESTful API

Pydantic

FastAPI relies heavily on Pydantic for manipulating (reading and returning) data.

In the root folder, create a new Python file called schema.py:

# schema.py

from pydantic import BaseModel

from typing import Optional


class UserBase(BaseModel):
    name: str
    email: str
    address: Optional[str] = None
    phone_number: Optional[str] = None
    sex: Optional[str] = None


class UserCreate(UserBase):
    email: str

class UserResult(UserBase):
    id: int

    class Config:
        orm_mode = True

Here, we defined a base model for a User object, and then added two Pydantic Models, one to read data and the other for returning data from the API. We used the Optional type for nullable values.

You can read more about Pydantic Models here.

In the UserResult Pydantic class, we added a Config class and set orm_mode to True. This tells Pydantic not just to read the data as a dict but also as an object with attributes. So, you'll be able to do either:

user_id = user["id"]   # as a dict

user_id = user.id  # as an attribute

More on Pydantic ORM models.

Next, add Models for Post and Comment objects:

# schema.py

from pydantic import BaseModel

from typing import Optional


class UserBase(BaseModel):
    name: str
    email: str
    address: Optional[str] = None
    phone_number: Optional[str] = None
    sex: Optional[str] = None


class UserCreate(UserBase):
    email: str


class UserResult(UserBase):
    id: int

    class Config:
        orm_mode = True


class PostBase(BaseModel):
    user_id: int
    title: str
    body: str


class PostCreate(PostBase):
    pass


class PostResult(PostBase):
    id: int

    class Config:
        orm_mode = True


class CommentBase(BaseModel):
    user_id: int
    body: str


class CommentCreate(CommentBase):
    pass


class CommentResult(CommentBase):
    id: int
    post_id: int

    class Config:
        orm_mode = True

API Endpoints

Now, let's add the API endpoints. In the main.py file, import in the Pydantic schema and the Masonite models:

import schema

from models.Post import Post
from models.User import User
from models.Comment import Comment

To get all users, we can use Masonite's .all method call on the User model on the collection instance returned:

@app.get("/api/v1/users", response_model=List[schema.UserResult])
def get_all_users():
    users = User.all()
    return users.all()

Make sure to import typing.List:

from typing import List

To add a user, add the following POST endpoint:

@app.post("/api/v1/users", response_model=schema.UserResult)
def add_user(user_data: schema.UserCreate):
    user = User.where("email", user_data.email).get()
    if user:
        raise HTTPException(status_code=400, detail="User already exists")
    user = User()
    user.email = user_data.email
    user.name = user_data.name
    user.address = user_data.address
    user.sex = user_data.sex
    user.phone_number = user_data.phone_number

    user.save() # saves user details to the database.
    return user

Import HTTPException:

from fastapi import FastAPI, HTTPException

Retrieve a single user:

@app.get("/api/v1/users/{user_id}", response_model=schema.UserResult)
def get_single_user(user_id: int):
    user = User.find(user_id)
    return user

Post endpoints:

@app.get("/api/v1/posts", response_model=List[schema.PostResult])
def get_all_posts():
    all_posts = Post.all()
    return all_posts.all()


@app.get("/api/v1/posts/{post_id}", response_model=schema.PostResult)
def get_single_post(post_id: int):
    post = Post.find(post_id)
    return post


@app.post("/api/v1/posts", response_model=schema.PostResult)
def add_new_post(post_data: schema.PostCreate):
    user = User.find(post_data.user_id)
    if not user:
        raise HTTPException(status_code=400, detail="User not found")
    post = Post()
    post.title = post_data.title
    post.body = post_data.body
    post.user_id = post_data.user_id
    post.save()

    user.attach("posts", post)

    return post

We saved the data from the API into the posts table on the database, and then in order to link the Post to the User, we attached it so that when we called user.posts(), we got all the user's posts.

Comment endpoints:

@app.post("/api/v1/{post_id}/comments", response_model=schema.CommentResult)
def add_new_comment(post_id: int, comment_data: schema.CommentCreate):
    post = Post.find(post_id)
    if not post:
        raise HTTPException(status_code=400, detail="Post not found")
    user = User.find(comment_data.user_id)
    if not user:
        raise HTTPException(status_code=400, detail="User not found")

    comment = Comment()
    comment.body = comment_data.body
    comment.user_id = comment_data.user_id
    comment.post_id = post_id

    comment.save()

    user.attach("comments", comment)
    post.attach("comments", comment)

    return comment


@app.get("/api/v1/posts/{post_id}/comments", response_model=List[schema.CommentResult])
def get_post_comments(post_id):
    post = Post.find(post_id)
    return post.comments.all()


@app.get("/api/v1/users/{user_id}/comments", response_model=List[schema.CommentResult])
def get_user_comments(user_id):
    user = User.find(user_id)
    return user.comments.all()

Start up the FastAPI server if it's not already running:

(.env)$ uvicorn main:app --reload

Navigate to http://localhost:8000/docs to view the Swagger/OpenAPI documentation of all the endpoints. Test each endpoint out to see the response.

Tests

Since we're good citizens, we'll add some tests.

Fixtures

Let's write tests for our code above. Since we'll be using pytest, go ahead and add the dependency to the requirements.txt file:

pytest==7.2.1

We would also need the HTTPX library since FastAPI's TestClient is based on it. Add it to the requirements file as well:

httpx==0.23.3

Install:

(.env)$ pip install -r requirements.txt

Next, let's create a separate config file for our tests to use so we don't overwrite data in our main development database. Inside the "config" folder, create a new file called test_config.py:

# config/test_config.py

from masoniteorm.connections import ConnectionResolver


DATABASES = {
  "default": "sqlite",
  "sqlite": {
    "driver": "sqlite",
    "database": "db.sqlite3",
  }
}

DB = ConnectionResolver().set_connection_details(DATABASES)

Notice that it's similar to what we have in the config/database.py file. The only difference is that we set our default to be sqlite as we want to use SQLite for testing.

In order to set our test suites to always make use of our config in the test_config.py configuration instead of the default database.py file, we can make use of pytest's autouse fixture.

Create a new folder called "tests", and within that new folder create a conftest.py file:

import pytest
from masoniteorm.migrations import Migration


@pytest.fixture(autouse=True)
def setup_database():
    config_path = "config/test_config.py"

    migrator = Migration(config_path=config_path)
    migrator.create_table_if_not_exists()

    migrator.refresh()

Here, we set Masonite's migration configuration path to the config/test_config.py file, created the migration table if it has not already been created before, and then refreshed all the migrations. So, every test will start with a clean copy of the database.

Now, let's define some fixtures for a user, post, and comment:

import pytest
from masoniteorm.migrations import Migration

from models.Comment import Comment
from models.Post import Post
from models.User import User


@pytest.fixture(autouse=True)
def setup_database():
    config_path = "config/test_config.py"

    migrator = Migration(config_path=config_path)
    migrator.create_table_if_not_exists()

    migrator.refresh()


@pytest.fixture(scope="function")
def user():
    user = User()
    user.name = "John Doe"
    user.address = "United States of Nigeria"
    user.phone_number = 123456789
    user.sex = "male"
    user.email = "[email protected]"
    user.save()

    return user


@pytest.fixture(scope="function")
def post(user):
    post = Post()
    post.title = "Test Title"
    post.body = "this is the post body and can be as long as possible"
    post.user_id = user.id
    post.save()

    user.attach("posts", post)
    return post


@pytest.fixture(scope="function")
def comment(user, post):
    comment = Comment()
    comment.body = "This is a comment body"
    comment.user_id = user.id
    comment.post_id = post.id

    comment.save()

    user.attach("comments", comment)
    post.attach("comments", comment)

    return comment

With that, we can now start writing some tests.

Test Specs

Create a new test file in the "tests" folder called test_views.py.

Start by adding the following to instantiate a TestClient:

from fastapi.testclient import TestClient

from main import app  # => FastAPI app created in our main.py file

client = TestClient(app)

Now, we'll add tests to:

  1. Save a user
  2. Get all users
  3. Get a single user using the user's ID

Code:

from fastapi.testclient import TestClient

from main import app  # => FastAPI app created in our main.py file
from models.User import User

client = TestClient(app)


def test_create_new_user():
    assert len(User.all()) == 0 # Asserting that there's no user in the database

    payload = {
        "name": "My name",
        "email": "[email protected]",
        "address": "My full Address",
        "sex": "male",
        "phone_number": 123456789
    }

    response = client.post("/api/v1/users", json=payload)
    assert response.status_code == 200

    assert len(User.all()) == 1


def test_get_all_user_details(user):
    response = client.get("/api/v1/users")
    assert response.status_code == 200

    result = response.json()
    assert type(result) is list

    assert len(result) == 1
    assert result[0]["name"] == user.name
    assert result[0]["email"] == user.email
    assert result[0]["id"] == user.id


# Test to get a single user
def test_get_single_user(user):
    response = client.get(f"/api/v1/users/{user.id}")
    assert response.status_code == 200

    result = response.json()
    assert type(result) is dict

    assert result["name"] == user.name
    assert result["email"] == user.email
    assert result["id"] == user.id

Run the tests to ensure they pass:

(.env)$ python -m pytest

Try writing tests for the post and comment views as well.

Conclusion

In this tutorial, we covered how to use the Masonite ORM together with FastAPI. The Masonite ORM is a relatively new ORM library with an active community. If you have experience with the Orator ORM (or any other Python-based ORM, for that matter), the Masonite ORM should be a breeze to use.

Featured Course

Test-Driven Development with FastAPI and Docker

In this course, you'll learn how to build, test, and deploy a text summarization service with Python, FastAPI, and Docker. The service itself will be exposed via a RESTful API and deployed to Heroku with Docker.

Buy Now $30 View Course
Featured Course

Test-Driven Development with FastAPI and Docker

In this course, you'll learn how to build, test, and deploy a text summarization service with Python, FastAPI, and Docker. The service itself will be exposed via a RESTful API and deployed to Heroku with Docker.

Buy Now $30 View Course