Building A Blog Service With FastAPI And Redis OM

Building A Blog Service With FastAPI And Redis OM

Introduction

In my last article, I wrote about using Redis as a message broker to store messages delivered by a task queue. Asides from Redis being used as cache storage or message broker, Redis can also be used as a database to store data in key-value pairs.

In this article, We will explore RedisJson and RedisSearch capabilities by building a blog service to create, retrieve, update, and carry out a full-text search using the Redis-OM and python's fastest web framework - FastAPI.

Objective

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

  • Create a Redis database on Redis
  • Create Database models with Redis-OM
  • Develop Restful APIs with Python FastAPI
  • Interact with Redis to create, retrieve and search data

Environment SetUp

Create a new folder on your desktop and name it "blog-fast-API"

$ mkdir blog-fast-API

$ cd blog-fastAPI

Create a virtual environment and activate it

For Windows users

$ python3  -m venv env 
$ cd env 
$ cd Scripts 
$ activate

For macOS users

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

in your root directory, create a folder named blog and create two new files under the blog called models.py and main.py. Also, create a .gitignore file and add env to ignore your environment variable when pushing to Github.

your directory should look like this :

Screenshot 2022-08-11 at 12.52.27.png

Install your dependencies by running

$ pip install fastapi
$ pip install redis-om
$ pip install "uvicorn[standard]"

FastAPI is a modern, high-performance Web framework for developing RESTful APIs in Python. it is fast, easy and automatically generates swagger API docs.

Run pip freeze > requirements.txt to ensure all your dependencies are in your environment.

in the blog/main.py file, let's create a root route to run the server.

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello world"}

Go to the blog directory and run the server

$ cd blog
$ uvicorn main:app --reload

Navigate to http://127.0.0.1:8000 in your browser, and you should see :

{"message":"Hello world"}

FastAPI generates swagger docs automatically for APIs. You can view your API docs by navigating to 127.0.0.1:8000/docs in your browser.

Screenshot 2022-08-11 at 13.09.55.png

Database Setup

We will be creating a new database on the Redis cloud.

Create an account on Redis and log in.

Navigate to the dashboard to create a new subscription. You can start with the free tier and upgrade later depending on your app usage.

Screenshot 2022-08-11 at 13.57.13.png

Create a new database and activate it, I will be calling mine "blog".

Screenshot 2022-08-11 at 14.01.38.png

in your blog/models.py file, import get_redis_connection and connect to your Redis database.

from redis_om import get_redis_connection

redis = get_redis_connection(
    host="redis-14531.c13.us-east-1-3.ec2.cloud.redislabs.com",
    port=14531,
    password="TbuxEUkupHczh3G5SVYFr2tF1SPmILDk",
    decode_responses=True,
)

host is the base URL on the public endpoint gotten from your database configuration on the dashboard, it stops at the .com.

port is the last five digits on your public endpoint

password is the default user password from your database configuration on the dashboard.

Creating Models

We will be creating two models for our app, the Author model and the Blog model. The Author model will be nested into the Blog Model.

in your 'blog/models.py' import EmbeddedJsonModel and JsonModel and create your models

import datetime

from redis_om import get_redis_connection, EmbeddedJsonModel, JsonModel, Field, Migrator

redis = get_redis_connection(
    host="redis-14531.c13.us-east-1-3.ec2.cloud.redislabs.com",
    port=14531,
    password="TbuxEUkupHczh3G5SVYFr2tF1SPmILDk",
    decode_responses=True,
)


class Author(EmbeddedJsonModel):
    first_name: str = Field(index=True, full_text_search=True)
    last_name: str
    email: str
    bio: str
    date_joined: datetime.date = Field(default=datetime.datetime.now())

    class Meta:
        database = redis


class Blog(JsonModel):
    title: str = Field(index=True, full_text_search=True)
    content: str
    author: Author
    date_posted: datetime.date = Field(
        default=datetime.datetime.today().strftime("%Y-%m-%d")
    )

    class Meta:
        database = redis


Migrator().run()

In the code above, We defined the schema that represents how my data will be stored in the Redis database.

The EmbeddedJsonModel class allows nesting of the Author model in the Blog model. The Field allows defining extra functions. The index=True tells Redis-OM we want to index first_name and title fields. The full_text_search=True tells Redis-OM we want to perform full_text_search query. The default sets a default date in a year/month/day format. The Migrator().run() runs migrations to set up indexes that Redis OM will use before we start querying.

Advanced Features of Redis OM like nesting JSON models inside each other and performing rich query expressions rely on two Redis modules: RedisJson and RedisSearch. RedisJson allows us to add JSON datatype with Redis while RedisSearch allows us to perform querying, indexing, and full_text_search with Redis.

Creating CRUD APIs

We will be creating the APIs to run CRUD operations.

in the main.py file, import the models and start creating the APIs

from fastapi import FastAPI
from models import Author, Blog

from redis_om.model import NotFoundError

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello world"}


# create an author
@app.post("/authors")
async def create_author(body: Author):
    author = Author(
        first_name=body.first_name,
        last_name=body.last_name,
        email=body.email,
        bio=body.bio,
        date_joined=body.date_joined,
    )

    author.save()

    return author


# create a blog
@app.post("/blogs")
async def create_blog(body: dict):
    author = Author.get(body["author_id"])
    blog = Blog(title=body["title"], content=body["content"], author=author)

    blog.save()

    return blog


# get a blog
@app.get("/blogs/{pk}")
async def get_blog(pk: str):
    try:
        blog = Blog.get(pk)
    except NotFoundError:
        return {"error": "no resource found"}, 404
    return blog


# update a blog
@app.put("/blogs/{pk}")
async def update_blog(pk: str, body: dict):
    blog = Blog.get(pk)

    blog.title = body["title"]
    blog.content = body["content"]

    blog.save()

    return blog


# delete a blog
@app.delete("/blogs/{pk}")
async def delete_blog(pk: str):
    Blog.delete(pk)
    return {"success": "blog deleted successfully"}

In the code above, we added basic endpoints to create an author, create a blog post, get a blog post, update a blog post, and delete a blog post.

The create a blog post takes the author_id as a request body to query out the author and embeds it while creating a new blog post. We also handled errors in case a blog post isn't found by importing the NotFoundError class provided by Redis OM.

Testing CRUD APIs

Now let's test the APIs we created above.

Go to 127.0.0.1:8000/docs in your browser to see the Swagger docs generated by FastAPI.

Screenshot 2022-08-11 at 18.37.01.png

Create Author

Screenshot 2022-08-11 at 20.52.00.png

Save the author pk somewhere to input when creating a new Blog.

Create Blog

Screenshot 2022-08-11 at 20.56.15.png

From the response above, the author details were nested into the blog response. This is one of the features of RedisJson.

Get a blog

Let's pass the pk of the blog created earlier and test the endpoint.

Screenshot 2022-08-11 at 21.01.02.png

Update a blog

Let's update the created blog by passing the pk

Screenshot 2022-08-11 at 21.06.23.png

Delete a blog

Let's pass the pk of the blog to test the endpoint

Screenshot 2022-08-11 at 21.08.50.png

Creating Search APIs

in blog/main.py add the following codes

def format_results(data):
    response = []
    for dat in data:
        response.append(dat.dict())

    return {"results": response}


@app.post("/blogs/find")
async def blog_by_name(title: str):
    blogs = Blog.find(Blog.title % title).all()

    return format_results(blogs)


@app.post("/blogs/find/author")
async def blog_by_author(first_name: str):
    blogs = Blog.find(Blog.author.first_name == first_name).all()

    return format_results(blogs)

In the code above, we created two APIs to search blog posts in the database. The blog_by_name API searches for blog posts whose title contains a full-text search that matches the search term. The blog_by_author API returns blog posts by the author's first_name passed. We also created a function to format the response returned by these APIs.

Now, let's test the APIs

Testing the Search APIs

Find by Name

Create different blog posts with a word similar in their title for a test.

Let's enter a keyword common to the blogs created to filter out blogs that match the keyword.

Screenshot 2022-08-11 at 21.37.51.png

Find by Author

Let's enter a created author's first_name to filter out all blog posts by that author.

Screenshot 2022-08-11 at 21.46.44.png

Conclusion

In this tutorial, you have learned how to use FastAPI and Redis OM to create APIs that perform crud operations. You have explored RedisJson capabilities by storing JSON documents and retrieving them. You have also used RedisSearch features to perform queries and full_text_search.

The codes can be found in this repository.

This post is in collaboration with Redis.

Learn more:

Did you find this article valuable?

Support Ubaydah Abdulwasiu by becoming a sponsor. Any amount is appreciated!