In this article, we will demonstrate how to build a data layer in Python for Azure Cosmos DB using FastAPI, a web framework for building and testing Python APIs.

To demonstrate this, we will leverage the To Do App Quickstart in our docs. This app shows simple CRUD operations (insert, update, list and delete) using a simple data model for a To-Do item consisting of id, name, description, and isComplete fields. The application that we will create, will wrap these CRUD operations and expose them through a set of custom APIs (Application Programming Interfaces) using the FastAPI routing feature. We will then test our custom APIs in Swagger UI that comes pre-packaged with FastAPI.

Here’s a complete list of steps we are going to follow to create and test our new backend data access layer.:

1. Create an Azure Cosmos DB account
2. Setup the dev environment
3. Creating .env file for credentials
4. Connect to account, create database and container
5. Define our ToDoItem data model
6. Setup the API endpoint implementation
7. Write our CRUD operations
8. Register the API endpoints with FastAPI
9. Test the endpoints

Step 1 – Create an Azure Cosmos DB account

In a new web browser window or tab, navigate to the Azure portal (portal.azure.com).

1. Sign-in into the Azure Portal using the credentials associated with your subscription.
2. Select Create a resource, search for Azure Cosmos DB, and then create a new Azure Cosmos DB SQL API account resource with the following settings, leaving all remaining settings to their default values:

   Setting	Value
   Subscription	Your existing Azure subscription
   Resource group	Select an existing or create a new resource group
   Account Name	Enter a globally unique account name
   Location	Choose any available region
   Capacity mode	Serverless

   Below is a screenshot showing these settings in the Azure Portal, click Review+Create and follow subsequent process.

   

    

3. Wait for the deployment task to complete before continuing with further steps.
4. Go to the newly created Azure Cosmos DB account resource and navigate to the Keys pane.
5. The “Keys” section contains the connection details and credentials necessary to connect to the account from the SDK. Specifically:
   • Record the value of the URI field. You will use this endpoint value later in this exercise.
   • Record the value of the PRIMARY KEY field. You will use this key value later in this exercise.
6. Close your web browser window or tab.

Step 2 – Setup the dev environment

For this application, you can use any terminal for executing your python and other commands. We’ve used VS Code as our editor here, but any editor works here. We will install FastAPI and Azure Cosmos DB Python SDK library following the steps below:

1. If Python is already installed, confirm it is version 3.6 or higher by executing python –version in the Terminal. If not the correct version or if not installed, install Python 3.6+ version, then execute python –version to verify correct installation.
2. Installing the remaining python libraries
   1. Fast API – Use pip install fastapi to install Fast API
   2. Uvicorn – Uvicorn is an ASGI web server implementation for Python. Use the command pip install fastapi uvicorn to install uvicorn
   3. Dotenv – Python-dotenv reads key-value pairs from a .env file and can set them as environment variables. Use pip install python-dotenv to install dotenv library
   4. Aiohttp – Asynchronous HTTP Client/Server for asyncio and Python.. Use pip install aiohttp  to install aiohttp library
   5. Azure Cosmos DB python client – Use pip install azure-cosmos to install azure-cosmos-db python client library.
3. You’ll need an editor to write code. We will use Visual Studio Code (VS Code) but you can use any code editor you’re comfortable with for writing Python.
4. Create a directory to store our code and related assets. For example, cosmos-fast. Then navigate to it in your terminal.
5. Open a terminal
6. Use cd <path to cosmos-fast> to navigate to cosmos-fast directory in the terminal/command prompt e.g., cd "C:\Cosmos\python-todo\cosmos-fast"

Step 3 – Creating .env file for credentials

To execute CRUD operations, we need to connect to the Azure Cosmos DB account and create a database and collection for our ToDo app. To connect to Azure Cosmos DB, we need the URI and PRIMARY KEY from Step 1. Follow the below steps:

1. Create and open a file .env in the Code Editor
2. Write your Azure Cosmos DB URI and KEY values to the .env file and save it

Step 4 – Connect to Azure Cosmos DB, create database and container

To connect to our Azure Cosmos DB account, use the credentials stored in the .env file. In the below snippet, we read the URI and KEY using the dotenv library and use these to instantiate the Azure Cosmos DB client during the app start. The Azure Cosmos DB async client is a part of the azure.cosmos.aio package. We will also import PartitionKey and exceptions from the azure.cosmos package. FastAPI will be used to handle the app startup event when the app loads on the server and later it would be used for request routing. Follow the steps below:

1. • Create main.py file and open it in the Code Editor.
   • Add the necessary imports at the top of the main.py file as shown below:

     from fastapi import FastAPI
     from dotenv import dotenv_values
     from azure.cosmos.aio import CosmosClient
     from azure.cosmos import PartitionKey, exceptions
     

   • Load the credentials in config variable, instantiate FastAPI, and define the Database and Container name.

     config = dotenv_values(".env")
     app = FastAPI()
     DATABASE_NAME = "todo-db"
     CONTAINER_NAME = "todo-items"

   • Connect to the Azure Cosmos DB account by instantiating the Cosmos Client during the app startup event, store a reference to the database and container in the app object to use them later, and then close the client connection.

     @app.on_event("startup")
     async def startup_db_client():
          app.cosmos_client = CosmosClient(config["URI"], credential = config["KEY"])
          await get_or_create_db(DATABASE_NAME)
          await get_or_create_container(CONTAINER_NAME)

• • Define the async get_or_create_db and get_or_create_container functions to fetch/create the database and the container.

    async def get_or_create_db(db_name):
        try:
            app.database  = app.cosmos_client.get_database_client(db_name)
            return await app.database.read()
        except exceptions.CosmosResourceNotFoundError:
            print("Creating database")
            return await app.cosmos_client.create_database(db_name)
         
    async def get_or_create_container(container_name):
        try:        
            app.todo_items_container = app.database.get_container_client(container_name)
            return await app.todo_items_container.read()   
        except exceptions.CosmosResourceNotFoundError:
            print("Creating container with id as partition key")
            return await app.database.create_container(id=container_name, partition_key=PartitionKey(path="/id"))
        except exceptions.CosmosHttpResponseError:
            raise

    Important Note On Partition Key

    In the above snippet, please make a special note of the term ‘Partition Key’. Partition Key is an important concept in Azure Cosmos DB, and a good understanding of partition key best practices can help you in better data modelling for optimum utilization of your requests budget while minimizing throttling.
  • Launch a new terminal window and navigate to the cosmos-fast directory by using the cd command
  • Now start the server that will execute your python script using the uvicorn web server by typing the below command in the Terminal

    uvicorn main:app --reload

  • Notice the output, you must see the “Creating Database” message when you run it for the first time

Step 5 – Define our ToDoItem data model

Next, define the data model for our To-Do items. It is recommended to define and use the models to make sure that we can validate that we would be storing the correct form of data in our database. FastAPI handles a lot of the tasks under the hood and takes care of data validation, mapping data types, generating API documentation, etc.

For our use case, each Todo item will have an Id (Azure Cosmos DB requires us to provide a unique id for each item that we create), name, description, and isComplete field. id, name, and description are string datatypes. isComplete is a boolean datatype. This is our model below:

1. Create a models.py file and open it in your Code Editor.
2. In models.py, add required imports

   from pydantic import BaseModel
   from typing import Optional

3. Define the ToDoItem Class

   class ToDoItem(BaseModel):
       id : str 
       name : Optional[str]
       description : Optional[str]
       is_complete : Optional[bool]

4. Save the models.py file. At this point, the models.py file would look like the file below:

Step 6 – Setup the API endpoint implementation

Now it is time to write functions to create ToDo items in our container, fetch a list of all the ToDo items, update an existing item, and delete an item. When we make create, delete, update, fetch, or other requests from the application frontend, we need to expose the respective functions through some API endpoints so that the frontend can call these functions that are hosted on our servers.

To implement the API endpoints, we will create another file routes.py and later import it in main.py. We will make use of the APIRouter component of the FastAPI for implementing the routes. Follow the steps below:

1. Create routes.py file in the same folder as the main.py, and open it in the Code Editor.
2. In the routes.py file, add the following imports

   from fastapi import APIRouter, Request
   from fastapi.encoders import jsonable_encoder
   from typing import List

3. Now, we also need to import our ToDoItem model defined in models.py file as we would use the ToDoItem model in our requests and responses. Add the below line to import from models

   from models import ToDoItem

4. We also need to instantiate APIRouter() from FastAPI to define the paths to access the defined functions. Add the below line to instantiate APIRouter:

   router = APIRouter()

5. Save the file and move on to the next step where we discuss various endpoints.

Step 7 – API endpoint implementation

Next, we will implement the four different API endpoints around the four CRUD operations mentioned earlier. This includes a POST request to CREATE a Todo item, a GET request to FETCH a list of all Todo items, a PATCH request to update a Todo item, and a DELETE request to delete a ToDoItem. These operations will also implement json_encoder. Json_encoder converts the ToDoItem object to a json dict.

So let us start by creating an endpoint to insert a ToDo item into our container.

Step 7.1 – POST request to insert one ToDoItem

Copy and add the code below into your routes.py file.

@router.post("/insert", response_model=ToDoItem)
async def create_todo(request: Request, todo_item: ToDoItem):
    todo_item = jsonable_encoder(todo_item)
    new_todo = await request.app.todo_items_container.create_item(todo_item)
    return new_todo

Now let us understand it line by line.

1. The parameters inside the router.post bracket define the path at which the function is hosted, and the model to deserialize the resonse into.
2. Inside create_todo function, we use the jsonable_encoder function to convert the todo_item object into a json dictionary as the Azure Cosmos DB client library allows only json objects in create_item.
3. Next, we call the create_item function on our container in which we want to insert the ToDoItem.
4. A successful create_item returns the newly created item which is then passed as the function’s return value as new_todo.
5. Since we are using the Azure Cosmos DB Python Async SDK and we need to await on the create_item, the create_todo method must be declared as async.

Step 7.2 – GET Request to fetch a list of all ToDo Items

We will use the read_all_items function from the Azure Cosmos DB Python client library to get a list of all the ToDo items. Define function and expose it through a GET request as below:

@router.get("/listall", response_description="List of all To-do items", response_model=List[ToDoItem])
async def list_todos(request: Request):
    todos = [todo async for todo in request.app.todo_items_container.read_all_items()]
    return todos

Step 7.3 – Delete an Item

Delete an item by specifying the id and the partition key. Define the delete function below that take id and partition key as the parameters. Define the function as below:

@router.delete("/delete")
async def delete_todo(request: Request, item_id: str, pk: str):
     await request.app.todo_items_container.delete_item(item_id, partition_key=pk)

Step 7.4 – Update an item

Update an item by passing the id and the item to be updated. Updates can be done on all properties except the partition key. In this case, that is id, which needs to be passed in order to update the correct item. In our example, we will use the same ToDoItem class to get the update fields in the request body. We will first read the item, update our properties, then call replace_item to replace the existing item with the modified item. Define the function as below:

@router.put("/update", response_model = ToDoItem, )
async def replace_todo(request: Request, item_with_update:ToDoItem):
    ````
    Update an item. Id (which is also the PartitionKey in this case) values should reference the item to be updated:

    - **id**: [Mandatory] Old Item ID
    - **name**: [Optional] The new name.
    - **description**: [Optional] The new description
    - **isComplete**: [Optional] boolean  flag to mark a Todo complete or incomplete
    ````
    existing_item = await request.app.todo_items_container.read_item(item_with_update.id,partition_key = item_with_update.id)
    existing_item_dict = jsonable_encoder(existing_item)
    update_dict = jsonable_encoder(item_with_update)
    for (k) in update_dict.keys():
        if update_dict[k]:
            existing_item_dict[k] = update_dict[k]
    updatedItem = await request.app.todo_items_container.replace_item(item_with_update.id, existing_item_dict)    
    return updatedItem

Save the file.

Step 8 – Register the API endpoints with FastAPI

Next, register the routes for our API endpoints for our CRUD operations in main.py so the requests are routed to the correct paths.

In the main.py file,

1. Add router import

   from routes import router as todo_router

2. Add below line to include router app.include_router(todo_router, tags=[“todos”], prefix=”/todos”)
3. Save the file

Step 9 – Test the endpoints

FastAPI automatically generates your API documentation and lets you test it. Follow the below steps:

1. In your browser, go to http://localhost:8000/docs
2. You will see all your endpoints listed there as shown in the picture below:
3. Start testing the API endpoint by creating a ToDo Item.
   • Expand the Create Todo in the FastAPI page and click on Try It Out
   • Replace the “Request Body” content with the below JSON.

     {   
       “id”: “unique-item-id-1”,   
       “name”: “To Do Title 1”,  
       “description”: “Just a description of the todo”, 
       “isComplete”: false 
     }

4. Click on Execute and observe the response.
5. Just as we created an item, we can test the Delete, Update, and List operations by providing the required parameters and executing.
6. Keep in mind that you cannot modify the id field for any item as it is the Partition Key.

Conclusion

This post demonstrated how to create a python web app backend using Azure Cosmos DB Python SDK and FastAPI. We created API endpoints to expose database CRUD operations, which can in turn be used by an application’s frontend or other services. We also tested our API endpoints by using the built-in features of FastAPI. This article touched upon the basic concepts that will help you get started. In the subsequent articles, we will cover additional and advanced concepts, including best practices for making the most of the Azure Cosmos DB.

About Azure Cosmos DB

Azure Cosmos DB is a fast and scalable cloud database for modern app development. See how to get started, dev/test, and run small production workloads free.

Get product updates, ask questions, and learn more about Azure Cosmos DB by following us on LinkedIn, YouTube, and Twitter.