I have been exploring serverless databases for a while now so I decided to write an article exploring this concept. In this article, we will build a serverless application leveraging a serverless database (Fauna) and FastAPI, a Python framework for building APIs.

The application we would be building is a simple CRUD API with an authentication feature using FastAPI and Fauna.

Serverless Databases

Serverless databases allow developers to focus on the main application. They are always hosted on a different server hence we can deploy our application on Heroku and our database will be on a server like AWS. This makes a lot of work easier. For instance, if I build an application running locally, and for some reason, I have to deploy the application to NameCheap. To use a database on the same server, I would have to use MySQL. This means I have to rewrite the database code even if I am using an ORM because they are two different database types. But with serverless databases like Fauna, I don't have to worry about that.

Fauna is a NoSQL serverless database. One reason why I like Fauna is that it's easy to use and they have a very nice community. They have drivers for Python, JavaScript, Java, etc. If you use MongoDB without using an ODM you'd see that working with data is easier in Fauna.

Our API Structure

To get started create a folder called crud-fastapi-fauna and create a virtual environment inside it then activate. Next, run the command below in your terminal to complete the setup.

$ pip install fastapi uvicorn pyjwt faunadb bcrypt python-dotenv

• /users/create/ - HTTP POST - Creating a new user

• /users/token/ - HTTP GET - Getting a token for authenticating users

• /users/todos/ - HTTP POST - Creating a new todo for the authenticated user

• /users/todos/ - HTTP GET - Getting all todos created by the authenticated user

• /users/todos/{todo id}/ - HTTP GET - Getting the todo data with the specified id. The authenticated user must be the one that created the todo.

• /users/todos/{todo id}/ - HTTP PUT - Updating the todo data whose id is given in the path parameter. The authenticated user must be the one that created the todo.

• /users/todos/{todo id}/ - HTTP DELETE - Deleting the todo data whose id is specified as a path parameter. The authenticated user must be the one that created the todo.

Getting Started with Fauna

To use Fauna, you need to create an account first. We also need to create a database and collection. We can do these in two ways. We can either create them directly on the website or use the Python driver to create them in our code. I'll show how to do it using both techniques in this article.

We also need to create an index. If you're from another NoSQL language like MongoDB you should be familiar with this. It's just a way we can search through documents(row) in our collection(table). We can create unique indexes like primary keys in SQL and also non-unique indexes that will return an array of data based on the condition. An index can also be created with both the driver(in this case the Python driver) and also directly on the website.

Creating a database

Navigate to the database section and create a new database. Prepopulating the database with demo data means you want to have sample data. In our case, we don't want that.

Creating Collection

To create a collection, navigate to the collection section in your dashboard and follow the put in the collection name. Similarly, create a todos collection. The most important thing is the name, you can leave other fields as it is.

Creating an Index

An index is what is used to query a collection in NoSQL databases. It can be unique like a primary key in SQL databases or a non-unique index. In this article I'll be creating 4 indexes which are user_by_id, user_by_email, todo_by_id, and todo_by_user_id. The first three have to be unique, but the last one may not be. This is because a user can have more than one todo. To create a new index, navigate to the index section in your dashboard and click “new index”. Note the terms which show the field that we want to index, we use data.id as the field name this is because when creating a new document we wrap it in a dictionary whose key is data.

Creating a secret key in Fauna

A secret key is what is used to communicate to a Fauna database from a Fauna driver. For example, while trying to communicate with Fauna using Python, we need to tell Fauna that this is who we are and this is the database we are trying to connect to, we do this through a secret key. To generate a secret key, navigate to the security section of your dashboard and click the “new key” button. You need to store the key immediately as Fauna auto generates a new key for you every time you click on the new key.

Working with Fauna’s Python Driver

Even though we created our database, collections, indices, and keys from the website we can also create this using code. Let’s see how we can achieve this:

Creating a Database

client.query(q.create_database({"name": "test"}))

We can’t access the database yet and that’s because there’s no data inside it. We need to generate a secret key to access the database. A key can have one of two different roles: server or admin. Let’s create a server key

client.query(
    q.create_key({
        "database": q.database("test"), 
        "role": "server"
    })
)

This returns a dictionary. What we need most is the secret. You can use this to create collections in the database. Note that this is a sub-database of the database we created on the website earlier.

Creating a Collection

server_client = FaunaClient(secret="your server key")
server_client.query(q.create_collection({"name": "test_coll"})) 

This creates a collection inside your database

server_client.query(
    q.create_index({
        "name": "id_index", 
        "source": q.collection("test_coll"),
        "terms": [{"field": ["data", "id"]}] 
    })
)

Getting Started with FastAPI

FastAPI is a python framework built on top of another python framework called Starlette. It uses an ASGI server called uvicorn to run, unlike Flask and Django which uses a WSGI server by default hence making FastAPI one of the fastest Python frameworks. One main reason why I love FastAPI a lot is that it auto-generates swagger documentation for us, and documentation and tests are two things developers don't really like writing. This documentation is generated for us due to some validations we did use type hinting. Consider the code snippet below

from fastapi import FastAPI 
from typing import List, Dict 

app = FastAPI() 

@app.get('/users/{user_id}', response_model=List[Dict[str, str]])
def get_user(user_id: str):
    return [
        {'name': 'Babatunde', 'id': user_id}
    ]

We first need to import the FastAPI class from fastAPI and we also need to import the List and Dict class from the typing module which is used for type hinting. There are 4 main decorators of fastAPI that are commonly used which are @app.get, @app.put, @app.post, and @app.delete which are used for HTTP GET, PUT, POST, and DELETE request respectively. These decorators take some parameters some of which are:

1. path: this is the endpoint that the view function handles the request for.

2. response_model: this is how the data returned should look like. This is great because we can receive data from the database that contains extra data that we don't want to pass, instead of looping and getting the needed data all I need to just do is use the response model and FastAPI helps in the data filtering. Here I did something very simple. In the later part of the article, I'll be relying on pydantic, which is another built-in python library for advanced type hinting.

3. status_code: This is the HTTP status the endpoint should return, default is 200(OK).

While defining our functions, we pass parameters which defines path parameters, query parameters, body, authorization data, etc

1. path parameters: must be of type string or int, and must be enclosed in the path of the decorator.

2. query parameter: must be of type int, str, or bool. Can also be defined with the Query class of FastAPI

3. Body parameters: Can be defined either using the Body class or using a class that inherits from pydantic.BaseModel

4. Header Parameters: Are defined using the Header class of FastAPI

To run the application type:

$ uvicorn main:app --reload

You should see something similar to this

The main:app denotes that uvicorn should run the app object in the main.py file in the current directory and --reload denotes that uvicorn should restart the server if it detects changes in the file. Go to http://127.0.0.1:8000/docs to see the documentation and test the API. You can also use postman for API testing.

As you can see, FastAPI generates swagger documentation for us and we can also see some default schema. If you click on the endpoint, you can test it out.

API Creation

To get started you need to create some files which are main.py, schema.py, models.py, .env and if you'll be committing your files to git you need to create a .gitignore file so your folder structure should be like this:

In your .env file put in your environment variables which are FAUNA_SECRET and SECRET_KEY. Fauna secret is the secret key given to you by Fauna while the secret key is a random key that we used while hashing user passwords.

FAUNA_SECRET='your fauna secret key'
SECRET_KEY='something secure'

In your schema.py file type the following:

'''Schema of all data received and sent back to the user'''
from pydantic import BaseModel, Field
from typing import Optional, List

class DeletedData(BaseModel):
    message: str 

class User(BaseModel):
    '''Base User schema contains name and email'''
    email: str = Field(
        None, title="The email of the user", max_length=300
    )
    fullname: str = Field(
        None, title="The name of the user", max_length=300
    )

class Todo(BaseModel):
    '''
    Schema of data expected when creating a new todo. Contains name and is_completed field
    '''
    name: str = Field(None, title='The name of the todo')
    is_completed: bool = Field(
        False, title='Determines if the todo is completed or not defaults to False'
    )

class UserInput(User):
    '''
    Schema of data expected when creating a new user. Contains name, email, and password
    '''
    password: str = Field(
        None, title="The password of the user", max_length=14, min_length=6
    )

class UserOutput(User):
    '''
    Schema of data returned when a new user is created. Contains name, email, and id
    '''
    id: str = Field(None, title='The unique id of the user', min_length=1)

class TodoWithId(Todo):
 '''Base schema of todo data returned when getting a todo data'''
    id: str 



class UserOutputWithTodo(UserOutput):
    '''
    Schema of data expected when getting all todo or when getting user data. 
    Contains name, email, id, and an array of todos
    '''
    todos: List[TodoWithId] = Field(
        [], title="The todos created by the user"
    )

class TodoOutput(TodoWithId):
    creator: UserOutput

class Token(BaseModel):
    token: str 

class UserSignin(BaseModel):
    email: str = Field(
        None, title="The email of the user", max_length=300
    )
    password: str = Field(
        None, title="The password of the user", max_length=300
    )

We defined some classes that rely on pydantic.BaseModel. These classes will help in data validation when receiving and sending data back to the client and that's all for this file.

Let's talk about the models.py file which is where I interact with Fauna. I love dividing my codes into classes if possible and I'd be doing the same here also. So I'll be creating two classes which are User and Todo. A user object should have full name, email, password, and id fields while a Todo data should have name, is_completed, id, and user_id fields. If you look closely in the schema.py file our schema for Todo and UserInput validates this except that there is no id and user_id there. But TodoOutput and UserOutputWithTodo have those fields except the password field since I don't want to expose the password of the user even though I'll be hashing it before storing it to the database.

In your models.py file type the following:

from faunadb import query as q
from faunadb.client import FaunaClient
from faunadb.objects import Ref
from faunadb.errors import BadRequest, NotFound
from dotenv import load_dotenv
from typing import Dict
import os, secrets

load_dotenv()

client = FaunaClient(secret=os.getenv('FAUNA_SECRET'))
indexes = client.query(q.paginate(q.indexes()))

print(indexes) # Returns an array of all index created for the database.

class User:

    def __init__(self) -> None:
        self.collection = q.collection('users')

    def create_user(self, data) -> Dict[str, str]:
        new_data = client.query(
            q.create(
                self.collection,
                {'data': {data, 'id': secrets.token_hex(12)}}
            )
        ) 
        return new_data['data']

    def get_user(self, id):
        try:
            user = client.query(
                q.get(q.match(q.index('user_by_id'), id))
            )
        except NotFound:
            return None
        return None if user.get('errors') else user['data']

    def get_user_by_email(self, email):
        try:
            user = client.query(
                q.get(q.match(q.index('user_by_email'), email))
            )
        except NotFound:
            return None
        return None if user.get('errors') else user['data']

class Todo:

    def __init__(self) -> None:
        self.collection = q.collection('todos')

    def create_todo(self, user_id, data) -> Dict[str, str]:
        new_todo = client.query(
            q.create(
                self.collection,
                {'data': {data, 'user_id': user_id, 'id': secrets.token_hex(12)}}
            )
        )

        return new_todo['data']

    def get_todo(self, id):
        try:
            todo = client.query(
                q.get(q.match(q.index('todo_by_id'), id))
            )
        except NotFound:
            return None
        return None if todo.get('errors') else todo 

    def get_todos(self, user_id):
        try:
            todos=client.query(q.paginate(q.match(q.index("todo_by_user_id"), user_id)))

            return [
                client.query(
                    q.get(q.ref(q.collection("todos"), todo.id()))
                )['data'] 
                for todo in todos['data']
            ] 
        except NotFound:
            return None 

    def update_todo(self, id, data):
    try:
        return client.query(
            q.update(
                q.ref(q.collection("todos"), id),
                {'data': data}
            )
        )['data']
    except NotFound: 
        return 'Not found'

    def delete_todo(self, id):
        try:
            return client.query(q.delete(q.ref(q.collection("todos"), id)))['data']
        except NotFound:
            return None

Firstly, I imported a bunch of stuff from Fauna.

1. FaunaClient: This is what we used to authenticate with Fauna by providing our secret key. There are other parameters but in this case, only the secret key is enough. If you have Fauna installed locally you need to specify domain, scheme, and port. The client.query takes in a query object.

2. Ref: all documents have a Ref object which is where the id generated by Fauna is stored.

3. NotFound: This is an error that is raised by Fauna if the document or collection is not found. There's also a bad request error that is raised if Fauna can't connect to the database.

We also got the indexes in our optional Fauna database.

The User class has three methods and one attribute.

1. The attribute collection: uses the query module to the users collection.

2. The create_user method takes the data passed from the client after being validated and passes it to this function, hence it has a full name, email and password. Then we used the secrets module to generate a unique identifier. The query object, q.create, takes in two parameters which are, the collection, and the data to be created as the document. The client.query method returns a dictionary containing ref and data as keys if no error was raised. The data is what we created and that's why it's what we're returning. And if an error was raised I returned None so in the view function we can send an error message like, “a bad request”.

3. The get_user takes in an id that we generated ourselves and uses the index we created earlier user_by_id to get the user. We use the q.get to query the index using q.match, which takes in the index object and the value we want to match in this case id. We can always use the id generated by Fauna but that means we have to add extra data after creating the user before returning it. But this will work perfectly well also.

4. The get_user_by_email takes in an email and uses the user_by_email index we created earlier to find the user.

I also ensured to catch document not found errors so as not to break our application.

The query module has other functions like query.update, query.replace, query.delete which updates, replaces, and deletes the document or whatever we want to work on.

The Todo class has one attribute which is used to define the collection and 5 methods which are create_todo, get_todo, get_todos, update_todo and delete_todo. These methods do something similar to the User class but the get_todos method is worth talking about. We use the index todo_by_user_id, this index is not unique since a user can have many todos. Firstly, I used the query.paginate which is used to return an array while querying then we pass the q.match into it so it can match all data with that user_id. This returns an array of ref objects in which we can use the ref object to get id and then get the todo data itself. Instead of using q.match I used q.ref, q.match is used when we're using an index, and q.ref is used when we're using a ref id. The ref object has an id method that returns the Fauna generated id.

What is remaining is integrating both the schema and models in our application.

Let's go to the main.py file and add the following code.

from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, status, Header
import models
import bcrypt, os, jwt 

from schema import 

load_dotenv()

app = FastAPI()

@app.post(
    '/users/create/', 
    response_model=UserOutputWithTodo, 
    description='This route is for creating user accounts',
    status_code=status.HTTP_201_CREATED
)
async def create_user(user: UserInput):
    user.password = bcrypt.hashpw(
        user.password.encode('utf-8'), 
        bcrypt.gensalt()
    ).decode('utf-8')

    try:
        user = models.User().create_user(user.dict())
    except Exception as e:
        raise HTTPException(400, detail=str(e))
    return user

We imported bcrypt which is what we'll be using to hash passwords while jwt is what we'll be using to generate an encoded token for authenticating the user.

I passed some parameters into the decorator, we've seen path, response_model, and status code before. The description is used to describe what the endpoint does and FastAPI will use it when describing the endpoint in swagger. Our Response Model is UserOutputWithTodo which is a class that FastAPI will turn to JSON. This class has a full name, email, id, and a todos array fields.

We get the JSON data sent from the client which contains the full name, email, and password. If the client sends invalid data FastAPI sends an error response back to the client. We hashed the password first then passed the user data to the User class in the models.py file. This returns a dictionary containing full name, email, password, and id. We then return the user. Since we've given the default value for todos that's what will be used. To return an error message we raise an HTTPException and pass the status code and the detail of the error.

To test this endpoint, I’ll use the swagger documentation generated for us. Run the server again and go to the documentation

As you can see, it gives us the kind of data we’re to input thanks to FastAPI. Click the try it out button and enter your details

Let's define an endpoint that sends a token that the client will be used for authentication.

@app.post(
    '/users/token/', 
    response_model=Token, 
    description='This route is for creating user accounts'
)
async def get_token(user: UserSignin):
    user_data = models.User().get_user_by_email(user.email)
    if user_data and bcrypt.checkpw(
        user.password.encode('utf-8'), 
        user_data['password'].encode('utf-8')
    ):
        token = jwt.encode({'user': user_data}, key=os.getenv('SECRET_KEY'))
        return {
        'token': token
        }

    header = {'WWW-Authenticate': 'Basic'}
    raise HTTPException(
        status.HTTP_400_BAD_REQUEST, 
        detail='Invalid email or password',
        headers=header
    )

If you test this endpoint, it returns a token if user pass their correct credentials hence it returns an error message.

Let's define a function that helps us authenticate. We can also use a decorator but to make things simple we'll be sticking to functions

We used the bearer form for authentication. If the client doesn't use this format we send a 401 error back to the client

async def authorize(authorization: str):
    if not authorization:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED, 
            detail='Token not passed'
        )
    if len(authorization.split(' ')) != 2 or\
            authorization.split(' ')[0] != 'Bearer':
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED, 
            detail='Invalid Token'
        )
    token = authorization.split(' ')[1]
    return jwt.decode(token, key=os.getenv('SECRET_KEY'), algorithms=[ 'HS256'])['user']

Finally let's write the code that handles creating, getting, updating, and deleting the todos

For each of the following functions, we used the authorize function to get the user making the request.

For the create_todo function, we got the user_id and passed it along with the JSON data passed and send it to the create_todo method of the models. Todo class.

For get_todo, update_todo, and delete_todo we first check if the id of the user making the request is the same as the user_id of the todo to ensure that a user can not work on todo that isn't created by them.

@app.post(
    '/users/todos/',
    response_model=TodoOutput, 
    status_code=status.HTTP_201_CREATED
)
async def create_todo(
    todo: Todo, 
    Authorization: str= Header(
        None, 
        description='Authorization is in form of Bearer <token> where token is given in the /users/token/ endpoint'
    )   
):
    user = await authorize(Authorization)
    try:
        todo = models.Todo().create_todo(user['id'], todo.dict())
    except Exception as e:
        raise HTTPException(400, detail=str(e))
    todo['creator'] = user
    return todo 

@app.get(
    '/users/todos/{todo_id}', 
    response_model=TodoOutput
)
async def get_todo(
    todo_id: str= Field(..., description='Id of the todo'),
    Authorization: str= Header(
        None, 
        description='Authorization is in form of Bearer <token> where token is given in the /users/token/ endpoint'
    )
):
    user = await authorize(Authorization)
    try:
        todo = models.Todo().get_todo(todo_id) 
    except Exception as e:
        raise HTTPException(400, detail=str(e))
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, 
            detail='No todo with that id'
        )
    todo = todo['data']
    if todo and todo['user_id'] != user['id']:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    todo.update({"creator": user})
    return todo

@app.get(
    '/users/todos/', 
    response_model= UserOutputWithTodo,
    description='Get all Todos'
)
async def get_all_todos( 
    Authorization: str= Header(
        None, 
        description='Authorization is in form of Bearer <token> '\
        'where token is given in the /users/token/ endpoint'
    )
):
    user = await authorize(Authorization)
    # get all todo
    try:
        todos = models.Todo().get_todos(user['id'])
    except Exception as e:
        raise HTTPException(400, detail=str(e))
    if not user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, 
            detail='User does not exist'
        )
    # add user data
    todos = [] if not todos else todos
    user.update({'todos': todos})
    return user

@app.put(
    '/users/todos/{todo_id}', 
    response_model=TodoOutput
)
async def update_todo(
    todo_id: str =Field(..., description='Id of the todo'),
    data: Todo = Body(...),
    Authorization: str= Header(
        None, 
        description='Authorization is in form of Bearer <token> where token is given in the /users/token/ endpoint'
    )
):
    user = await authorize(Authorization)
    try:
        todo = models.Todo().get_todo(todo_id 
        if todo['data']['user_id'] != user['id']:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
        todo = models.Todo().update_todo(todo['ref'].id(), data.dict())
    except Exception as e:
        raise HTTPException(400, detail=str(e))
    todo.update({"creator": user})
    return todo

@app.delete(
    '/users/todos/{todo_id}', 
    response_model=DeletedData
)
async def delete_todo(
    todo_id: str = Field(..., description='Id of the todo'),
    Authorization: str= Header(
        None, 
        description='Authorization is in form of Bearer <token> where token is given in the /users/token/ endpoint'
    )
):
    user = await authorize(Authorization)
    try:
        todo = models.Todo().get_todo(todo_id)
        if not todo:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, 
            detail='No todo with that id'
        )
        if todo['data']['user_id'] != user['id']: 
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
        todo = models.Todo().delete_todo(todo['ref'].value['id'])
    except Exception as e: 
        raise HTTPException(400, detail=str(e))
    return {'message': 'Todo Deleted successsfully'}

Conclusion

In this article, you’ve been able to build a fully functional todo application using two interesting technologies, FastAPI and Fauna. You’ve also learned how to secure your user data by hashing data and finally learned how you can authenticate users in an API. If you like you can deploy this to a hosting platform of your choice and interact with the API from a front end application like React js. The source code for this project is available on GitHub If you find this article interesting please do share with your friends and colleagues. You can reach out to me via Twitter if you have any questions.