Note: This is an article I wrote in 2021 and never published for three years. Some part of it might not work today but it still is helpful for anyone looking to go from knowing just the fundamentals of Python to building an end-to-end API in fastAPI.

Introduction

Hey Everyone! In this article, we will go through all the features of FastAPI needed to build an API for a Todo App. We will perform CRUD(create-read-update-delete) operations, add a database and authentication to our API.

We will start by understanding a few prerequisites for the FastAPI app, then we will go through a hello world app for FastAPI and learn its fundamentals, and then finally end by building a todo app with FastAPI.

The only prerequisite to this article is to have experience coding in Python.

The best way to go through the article is to open your code editor and code along with me. (Don't copy-paste)

So let us start!

What is FastAPI?

FastAPI is a modern, fast (high-performance) web framework for building APIs with Python 3.6+.

But wait, what is an API? (feel free to skip if you already know)

API stands for Application programming interface, it is a set of functions that performs a task and can be used by other developers to build their own software without caring about the implementation detail of the API.

One of the most famous analogies of an API is to think of an API as a waiter in a restaurant, when you go to a restaurant a waiter takes your order, delivers it to the kitchen, and then delivers the food.

You may watch this 3 minutes video that explains API in a precise way:

Why use FastAPI to build APIs?

The answer is simple: FastAPI is awesome!

• It is Fast: FastAPI is a high-performance web framework, on par with NodeJS and Go.

• It is Fast to code: Increase the speed to develop features by about 200% to 300%.

• It is Intuitive: Great editor support. Completion everywhere. Less time debugging.

• It is Easy: Easy to use and learn.

• It is Short: Minimize code duplication.

• It is Robust: Get production-ready code. With automatic interactive documentation.

• It is Standard-bases: Based on (and fully compatible with) the open standards for APIs.

FastAPI uses some of the best libraries available in the Python community to achieve the above features:

1. Pydantic: For data validation, serialization, and documentation.

2. Starlette: Starlette is a lightweight ASGI web framework/toolkit.

3. Uvicorn: Uvicorn is a lightning-fast ASGI server.

Don't worry if don't understand a few terms here, we will go through all this as needed.

Python Type Hints

Before starting development with FastAPI, we also need to make sure we are comfortable using types in Python.

(Feel free to skip this section if you are already familiar with type hints in Python)

Type hints are a special syntax that allows declaring the type of a variable.

Editors and tools use these types of hints to provide better support like auto-completion and error checks.

FastAPI is based on these types hints and uses them to provide:

• Data validation

• Serialization

• Documentation support.

Using type hints

Type hints are declared using a : after the variable name, see the following example:

def add(num1: int, num2: int):
    return num1 + num2

You can also use any standard python types: int, float, str, bool, bytes

Using the typing module

For data structures that can contain other values, like dict, list, set, tuple you can use the standard python module typing. These types are called generic types.

Here's an example:

from typing import List

def process_items(items: List[str]):
    for item in items:
        print(item)

List[str]: The type of the values that a list can contain is given in square brackets []. It is called the type parameter.

That means: "the variable items is a list, and each of the items in this list is an str".

Tuple and Set: For declaring the type of tuples and sets:

from typing import Set, Tuple

def process_items(items_t: Tuple[int, int, str], items_s: Set[bytes]):
    return items_t, items_s

This means:

• The variable items_t is a tuple with 3 items, an int, another int, and str.

  If you want to declare a tuple with variable length and homogeneous type, you can use literal ellipsis: Tuple[int, ...].

  If you want to declare a plain tuple: Tuple[Any, ...].

• The variable items_s is a set, and each of its items is of type bytes.

Dicts:

To define a dict, you pass 2 type parameters, separated by commas.

from typing import Dict

def process_items(prices: Dict[str, float]):
    for item_name, item_price in prices.items():
        print(item_name)
        print(item_price)

This means:

• The variable prices is a dict:

  • The keys of this dict are of type str.

  • The values of this dict are of type float.

Optional

You can also use Optional to declare that a variable has a type, like str, but that it is "optional", which means that it could also be None.

from typing import Optional

def say_hi(name: Optional[str] = None):
    if name is not None:
        print(f"Hey {name}!")
    else:
        print("Hello, World!")

Using Optional[str] instead of just str will let the editor help you detecting errors where you could be assuming that a value is always str when it could actually be None too.

Union

Union types allow combining types: Union[int, str] means either int or str.

Literal

Literal types can be used to indicate that the corresponding variable can have one of the provided literal value: MODE = Literal['r', 'rb', 'w', 'wb'].

Using classes as type hints

You can also declare a class as the type of a variable.

class Person:
    def __init__(self, name: str):
        self.name = name

def get_person_name(a_person: Person):
    return a_person.name

Note 📝

Since Python 3.9 you can use built-in collection types such as list and dict as generic types instead of importing the corresponding types (ex. List and Dict) from typing module.

def greet_all(names: list[str]) -> None:
    for name in names:
        print("Hello", name)
# The -> indicate return type although it can be inferred automatically

Tip 💡

Install the mypy library to get error checks when using type hints.

Install $ python3 -m pip install mypy.

Setup in VS Code: Open the command palette (Ctrl+Shift+P) and select the Python: Select Linter command and choose mypy. Also, make sure you have the pylance and the python extension installed in your vs code for the best experience.

Hello World with FastAPI

Now, we will create a Hello World app in FastAPI and then discuss a few fundamental features.

Let's start by creating a new project folder:

$ mkdir hello
$ cd hello
$ touch __init__.py # this will create an __init__.py file to make this folder a package

Now, create a virtual environment and activate it:

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

Time to Install FastAPI:

$ pip install fastapi[all]

Hello, World!

Now, create a file [main.py](http://main.py) in the current directory and type the following into it, don't worry we will go over each line.

# Snippet 1

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def hello():
    return {"message": "Hello, World!"}

• from fastapi import FastAPI

  We are importing the FastAPI class from the fastapi module, it will provide all the functionality for our API, and make our app a FastAPI app.

• app = FastAPI()

  This creates an instance of the FastAPI class. We will also refer to this instance when running our API using a server. Also, you can name it anything instead of the app.

• @app.get("/")

  This line creates a path operation using a decorator.

  A Path refers to the last part of the URL starting from the first /.

  So, in a URL like: [https://example.com/items/foo](https://example.com/items/foo), the path would be /items/foo.

  An operation refers to one of the HTTP methods:

  • POST: to create data

  • GET: to read data

  • PUT: to update data

  • DELETE: to delete data

  • and more...

You can communicate to each path using one (or more) of these "methods" in the HTTP protocol.

So, the @app.get("/") decorator (a "decorator" takes the function below and does something with it) tells FastAPI that the function right below is in charge of handling requests that go to:

• The path /.

• Using a get operation.

You can also use the other operations:

• @app.post()

• @app.put()

• @app.delete()

• The hello() function

  This is a path operation function, it is called by FastAPI whenever it receives a request to the URL / using a GET operation. You can use a normal function or an async function.

  This function is returning a python dictionary which will be automatically converted to a JSON object by FastAPI.

  • JSON stands for JavaScript Object Notation.

  • It is a lightweight format for storing and transporting data.

  • JSON is often used when data is sent from a server to a web page.

  • JSON is very similar to a Python dictionary, so if you haven't ever used JSON then don't worry just think of it as a slightly different representation of a python dictionary.

You can also return other data types like list, str, int, etc.

Running a FastAPI app

Let's run the above hello world app, open your terminal and type the following command:

$ uvicorn main:app --reload

This will start our app on [localhost:8000](http://localhost:8000).

main refers to our Python module and app refers to the FastAPI instance we created.

--reload is used to restart the server after code changes. Only use for development.

The Autogenerated Docs

FastAPI provides automatic documentation, go to [localhost:8000/docs](http://localhost:8000/docs) and you will be able to see all the path operations and try them out.

FastAPI Fundamentals

Now, we will create more path operations in the same hello world app and understand a few important concepts in FastAPI like path parameters, query parameters, request body, using Pydantic models, response models, and dependencies.

We will learn about Routers, Databases, and Security while creating the Todo API in the next Section.

Path Parameters

Path parameters are the variable parts of the URL path.

In [https://twitter.com/yuvraajsj18](https://twitter.com/yuvraajsj18), yuvraajsj18 is a path parameter. You can insert any other value: https://twitter.com/{username}.

Let's create our own path with a path parameter:

# Snippet 2

# ... code from snippet 1

items = [
    {
        "item_id": 0,
        "name": "Item 1",
        "description": "Description for item 1",
        "price": 50.25,
    },
    {
        "item_id": 1,
        "name": "Item 2",
        "description": "Description for item 2",
        "price": 20.00,
        "tax": 2.0,
    },
    {
        "item_id": 2,
        "name": "Item 3",
        "description": "Description for item 3",
        "price": 30.00,
        "tax": 1.5,
    },
]

**@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id < len(items):
        return items[item_id]

    return {"Error": f"item_id {item_id} not found."}**

First we have created a items list to mock a database.

Then we are creating a get path operation at items/{item_id}.

{item_id} is the placeholder for our path parameter.

Whatever is passed into the URL will be automatically converted, validated, and saved into the int variable item_id declared in the function signature: get_item(item_id: int).

We can then use it in the definition. Here we retrieve the item and return the corresponding item dictionary if it exists otherwise an error message.

Try running this using the uvicorn server and going to the docs path.

Also, try passing a string parameter in the URL like [localhost:8000/items/foo](http://localhost:8000/items/foo) instead of int. You will get the following error:

{"detail":[{"loc":["path","item_id"],"msg":"value is not a valid integer","type":"type_error.integer"}]}

Query Parameters

Query Parameters are the key-value pairs that are provided after the ? and separated by & in the URL.

In the URL: https://www.google.com/search?q=cats, q=cats is a query parameter, it tells google what "query" is to search.

In FastAPI, When you declare other function parameters that are not part of the path parameters, they are automatically interpreted as "query" parameters.

Let's continue with our file written above and add a path operation that accepts query parameters.

# Snippet 3

# ... code from snippet 2

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return items[skip : skip + limit]

Try going to the [localhost:8000/items/?skip=1&limit=3](http://localhost:8000/items/?skip=1&limit=3), you will get the items[1: 3] that is the following 2 items:

[{"item_id":1,"name":"Item 2","description":"Description for item 2","price":20.0,"tax":2.0},
{"item_id":2,"name":"Item 3","description":"Description for item 3","price":30.0,"tax":1.5}]

FastAPI receives these query parameters from the URL as string, convert to their declared type, int in this case and store them in the corresponding variable.

As query parameters are not a fixed part of a path, they can be optional and can have default values.

In the example above they have default values of skip=0 and limit=10. So, going to [localhost:8000/items](http://localhost:8000/items) is same as going to localhost:8000/items/?skip=0&limit=10.

You can also create optional query parameters using None, let's add a greet function to illustrate:

# Snippet 4
from typing import Optional
# ... code from snippet 3

@app.get("/greet")
async def greet(name: Optional[str]= None):
    if name:
        return {"message": f"Hello, {name}"}
    return {"message": "Hello, Anonymous!"}

FastAPI will know that name is optional because of the = None.

The Optional in Optional[str] is not used by FastAPI (FastAPI will only use the str part), but the Optional[str] will let your editor help you find errors in your code.

When you want to make a query parameter required, you can just not declare any default value.

Request Body

When you need to send data from a client (let's say, a browser) to your API, you send it as a request body.

A request body is data sent by the client to your API.

A response body is the data your API sends to the client.

Example: When you fill up a sign-up form on a platform, the data is sent to the back-end using the request body.

To declare a request body, we use Pydanticmodels.

Let's create a path operation that adds a new item to our list of items:

# Snippet 5

from pydantic import BaseModel

# ... code from snippet 4

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

# To receive request body, you should use one of: 
# POST (the more common), PUT, DELETE or PATCH.
@app.post("/items/")
async def create_item(item: Item):
    items.append({"item_id": len(items), **item.dict()})
    return items[-1]

The Item class is a pydantic model that defines the shape of an item. When an instance of Item is created, the attributes used will be validated, parsed, and stored in the instance.

FastAPI will receive the JSON object from the client, and provide us with the corresponding pydantic model instance, in this case, item.

We can define optional attributes by assigning None to them, as done for the description and tax.

We will then use the item instance to append a new item in the items list. **item.dict() is dictionary unpacking syntax in python. It will return the keyword-value pair. We are also adding an item_id and then finally returning the newly created item.

Try Creating an Item using the /docs path:

The new item is created but it will not persist once you restart your app, we will use databases in the Todo API section to make data persistent.

You can use body, path, query parameters together. FastAPI will recognize each of them and take the data from the correct place.

Adding additional information and validation to parameters

We can use Query, Path, Field, Body to provide an additional string or numeric validation for our parameters.

Let's add string validation to the greet function we have written before:

# Snippet 4 Modified
from fastapi import FastAPI, Query  # import Query
from typing import Optional

@app.get("/greet")
async def greet(
    **name: Optional[str] = Query(None, min_length=1, title="Name", alias="username")**
):
    if name:
        return {"message": f"Hello, {name}"}
    return {"message": "Hello, Anonymous!"}

The first value to Query provides the default value for the parameter. It is followed by other optional validations:

• min_length: int, specifies the minimum length of the parameter value.

• max_length: int, specifies the maximum length of the parameter value.

• regex: str, defines a regular expression that the parameter should match.

• alias: str, provides an alternate path for the operation, now greet can be called with /greet/?username=Yuvraj instead of /greet/?name=Yuvraj.

• title: str, provides title to the query in the docs, a metadata argument.

• description: str, describes the query in the docs, a metadata argument.

• deprecated: bool, specifies whether the parameter is deprecated or not in the docs, a metadata argument.

When you don't want to provide a default value, you can use Query(..., min_length=1). ... is a special python value called ellipsis.

Now, let's add some numeric validation to the read_items function written before:

# Snippet 2 Modified
from fastapi import FastAPI, **Path,** Query

@app.get("/items/{item_id}")
**async def get_item(item_id: int = Path(..., ge=0)):**
    if item_id < len(items):
        return items[item_id]

    return {"Error": f"item_id {item_id} not found."}

As a path parameter is always required, the first argument to Path should always be the... It can be followed by the metadata arguments or the numeric validation arguments:

• gt: greater than

• ge: greater than or equal

• lt: less than

• le: less than or equal

Both string and numeric validation and metadata arguments can be used with Query and Path.

There also exists Field to add similar validation to a Pydantic model field and Body to add validation when receiving a singular value from the client instead of a dictionary.

Here is the example of both Field and Body

# Snippet 5 modified

from fastapi import **Body**, FastAPI, **HTTPException** Path, Query
from pydantic import BaseModel, **Field**

class Item(BaseModel):
    name: str
    description: Optional[str] = Field(None, max_length=300)
    price: float = Field(..., gt=0, description="The price must be greater than zero")
    tax: Optional[float] = None

@app.post("/items/")
async def create_item(item: Item):
    items.append({"item_id": len(items), **item.dict()})
    return items[-1]

@app.put("/items/{item_id}")
async def update_item(item_id: int, add_price: float = Body(...)):
    if item_id >= len(items):
        **raise HTTPException(status_code=404, detail="Item not found")**

    item = Item(**items[item_id])
    item.price += add_price

    items[item_id] = item.dict()

    return items[item_id]

• update_item expect a single value for add_price.

Handling Exceptions

In the above code, in function update_item we saw the use of HTTPException.

There are many situations where you need to notify an error to a client that is using your API. We can do it with the help of HTTPException class.

HTTPException is a normal Python exception with additional data relevant for APIs.

Because it's a Python exception, you don't return it, you raise it.

This also means that if you are inside a utility function that you are calling inside of your path operation function, and you raise the HTTPException from inside of that utility function, it won't run the rest of the code in the path operation function, it will terminate that request right away and send the HTTP error from the HTTPException to the client.

Following are the status codes to specify the response types:

1. Informational responses (100–199)

2. Successful responses (200–299)

3. Redirects (300–399)

4. Client errors (400–499)

5. Server errors (500–599)

Response Model

We can declare the model used for the response with the response_model in any of the path operations.

Let's modify the create_item operation to include a response model:

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    items.append({"item_id": len(items), **item.dict()})
    return items[-1]

Now go to the /docs and see the response section of the create_item, you will see the Item model instead of the string that was shown before.

FastAPI will use this response_model to:

• Convert the output data to its type declaration.

• Validate the data.

• Add a JSON Schema for the response, in the OpenAPI path operation.

• Will be used by the automatic documentation systems.

• Will limit the output data to that of the model.

  This means now when you run this operation, the item_id won't be returned anymore. As it is not included in the Item model. You can create multiple models for the request (input) and response (output), we will see this in detail in the Todo API app.

Dependencies

Okay, one last topic before moving on to creating the Todo API with FastAPI and using databases, routers, and authentication.

Dependency injection means, that there is a way for our code to declare things that it requires to work and use: "dependencies".

FastAPI will take care of doing whatever is needed to provide your code with those needed dependencies.

This is very useful when you need to:

• Have shared logic (the same code logic again and again).

• Share database connections.

• Enforce security, authentication, role requirements, etc.

• And many other things...

All these, while minimizing code repetition.

Let's implement a dependency in our hello app:

Remember we created a read_items function that takes two parameters skip and limit. Now we will do the same for a users list.

users = [
    {
        "id": 0,
        "username": "user1",
        "hashed_password": "somecrypticpassword1",
    },
    {
        "id": 1,
        "username": "user2",
        "hashed_password": "somecrypticpassword2",
    },
    {
        "id": 2,
        "username": "user2",
        "hashed_password": "somecrypticpassword2",
    },
]

@app.get("/users/")
async def read_users(skip: int = 0, limit: int = 10):
    return users[skip : skip + limit]

Now we will use a dependency to factor out this common parameter.

async def common_parameters(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

See, a dependency is just a function. Here it is taking the same parameter as the path functions.

Now we will use this in our path functions:

from fastapi import Body, **Depends**, FastAPI, HTTPException, Path, Query

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return items[commons["skip"] : commons["skip"] + commons["limit"]]

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return users[commons["skip"] : commons["skip"] + commons["limit"]]

Whenever a new request arrives, FastAPI will take care of:

• Calling your dependency ("dependable") function with the correct parameters.

• Get the result from your function.

• Assign that result to the parameter in your path operation function.

Try running the app using /docs. It still functions the same way as before.

We can also use classes as a dependency:

class CommonQueryParams:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(**commons: CommonQueryParams = Depends(CommonQueryParams)**):
    return items[commons.skip : commons.skip + commons.limit]

@app.get("/users/")
async def read_users(**commons: CommonQueryParams = Depends(CommonQueryParams)**):
    return items[commons.skip : commons.skip + commons.limit]

• Using classes allows your editor to provide code completion.

• You can use the following shortcut to avoid typing the type twice when using class dependency:

    commons: CommonQueryParams = Depends()
  

Few things to remember while using dependencies:

• You can create dependencies that have sub-dependencies. This means one dependency can be dependent on another.

• If one of your dependencies is declared multiple times for the same path operation, for example, multiple dependencies have a common sub-dependency, FastAPI will know to call that sub-dependency only once per request.

• When you don't need the return value of a dependency, you can add a dependencies list in the path operation decorator instead of declaring it in the path operation function.

    @app.get("/items/", dependencies=[Depends(dependency1), Depends(dependency2)])
    async def read_items():
        return items
  

• You can add dependencies to all the path operations by creating global dependencies on the app:

    app = FastAPI(dependencies=[Depends(dependency1), Depends(dependency2)])
  

• You can perform extra steps after finishing the dependency using yield instead of return.

  When the Python yield statement is encountered, the program suspends function execution, saves its state, and returns the yielded value to the caller. Whereas return stops function execution completely.

  You could use this to create a database session and close it after finishing.

  Only the code before and including the yield statement is executed before sending a response:

    async def get_db():
        db = DBSession()
        try:
            yield db
        finally:
            db.close()
  

  Note: you cannot raise an HTTPException after the yield statement as the response has already been sent to the user.

We will now put all of this together and use more features like databases, auth, and routing to create a Todo API.

Building the Todo API

Let's start by creating a new project:

$ mkdir todo
$ cd todo
$ touch __init__.py

Then, create and activate a virtual environment:

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

Install FastAPI

$ pip install fastapi[all]

Now, before going any further, let's define the requirements of our API.

Requirements:

1. User registration.

2. User authentication.

3. Create a new to-do item.

4. Delete an existing to-do item.

5. Mark a to-do item as done/not done.

6. View all to-do items.

Create a directory structure as follows:

todo ├── app │ ├── crud

│ ├── db │ ├── models │ ├── routers │ └── schemas └── env

Authentication

We will use tools provided by FastAPI to build our authentication system using OAuth2.

OAuth2 is a specification that defines several ways to handle authentication and authorization.

Let's see what are the tools provided by FastAPI for authentication(you don't have to code this yet, we will utilize these tools in Todo API later):

from fastapi import Depends, FastAPI
from fastapi. security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}

When we run this and try out the /items operation we will be required to provide the username and password.

This is the password flow defined in OAuth2. In this flow, the API itself is handling the authentication.

This is how everything works in this flow:

• The user types the username and password in the frontend and hits Enter.

• The frontend (running in the user's browser) sends that username and password to a specific URL in our API (declared with tokenUrl="token").

• The API checks that username and password, and responds with a "token" (we haven't implemented any of this yet).

  • A "token" is just a string with some content that we can use later to verify this user.

  • Normally, a token is set to expire after some time.

    • So, the user will have to log in again at some point later.

    • And if the token is stolen, the risk is less. It is not like a permanent key that will work forever (in most cases).

• The front end stores that token temporarily somewhere.

• The user clicks in the frontend to go to another section of the frontend web app.

• The front end needs to fetch some more data from the API.

  • But it needs authentication for that specific endpoint.

  • So, to authenticate with our API, it sends a header Authorization with a value of Bearer plus the token.

    (a header is just another part of the request sent by the client. Like the body, as we have seen before).

  • If the token contains foobar, the content of the Authorization header would be Bearer foobar.

FastAPI's OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

We are creating an instance oauth2_scheme of the class OAuth2PasswordBearer by passing a URL that will receive the username and password. This parameter doesn't create the path operation, we will do it ourselves later.

The instance oauth2_scheme is a callable, so it can be used with Depends.

Usingoauth2_scheme

@app.get("/items")
async def read_items(**token: str = Depends(oauth2_scheme)**):
    return {"token": token}

This dependency will go and look in the request for that Authorization header, check if the value is Bearer plus some token, and return the token as str that will be assigned to token the path operation function.

If it doesn't see an Authorization header, or the value doesn't have a Bearer token, it will respond with a 401 status code error (UNAUTHORIZED) directly.

You don't even have to check if the token exists to return an error. You can be sure that if your function is executed, it will have str in that token.

UsingUser Model

Let's make the auth system useful by creating a User model. (Again, you don't have to code this right now, we will be creating a better version for Todo API later).

from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None

Let's create a dependency get_current_user:

def fake_decode_token(token: str):
    return User(
            username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
    )

**async def get_current_user(token: str = Depends(oauth2_scheme)):**
    user = fake_decode_token(token)
    return user

get_current_user is a dependency that has oauth2_scheme as a sub-dependency. We are also using a fake utility function for now to decode tokens.

Inject the current user

Now, we will use get_current_user dependency in a path operation:

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

Getting theusername and password from client:

OAuth2 specifies that when using the "password flow" the client/user must send a username and password fields as form data.

The fields should also be named like this, email and password won't work. Although we can show the fields to the final users in the frontend as we wish.

Also, database models can use other names too.

UsingOAuth2PasswordRequestForm

from fastapi.security import OAuth2PasswordBearer, **OAuth2PasswordRequestForm**

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "alice@example.com",
        "hashed_password": "fakehashedsecret2",
    },
}

class UserInDB(User):  # User is same as defined before
    hashed_password: str

def fake_hash_password(password: str):
    return "fakehashed" + password

**@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):**
    user_dict = fake_user_db.get(form_data.username)

    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    user = UserInDB(**user_dict)
    hashed_password = fake_hash_password(form_data.password)

    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    return {"access_token": user.username, "token_type": "bearer"}

The OAuth2PasswordRequestForm is a class dependency that provides username and password received as form data.

(Form data in fastAPI is received using Form. It's similar to Body or Query. But instead of application/json encoding it uses application/x-www-form-urlencoded encoding. The <form></form> tag in HTML uses this encoding. Although you don't have to worry about this much as FastAPI will provide you with data directly just like it did with the body).

Then, we are getting the user data from the fake database using the username field. If no such user is found we are raising an error.

Now, we put the data in the pydantic model UserInDB and check if the hash of the password field matches the user's hashed password. If the passwords don't match, we raise an error.

Password Hashing

"Hashing" means: converting some content (a password in this case) into a sequence of bytes (just a string) that looks like gibberish.

Whenever you pass the same content (the same password) you get the same gibberish.

But you cannot convert from the gibberish back to the password.

Why use password hashing?

If your database is stolen, the thief won't have your users' plaintext passwords, only the hashes.

Returning the token:

The response of the token endpoint must be a JSON object.

It should have a token_type. In our case, as we are using "Bearer" tokens, the token type should be "bearer".

And it should have an access_token, with a string containing our access token.

In this example, we are returning the username as a token that is not secure, we will implement a secure token in the Todo API.

Update the dependencies

We will handle an exception in the previously written get_current_user dependency:

from fastapi import Depends, FastAPI, HTTPException, **status**

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def fake_decode_token(token: str):
    user = get_user(fake_users_db, token)
    return user

**async def get_current_user(token: str = Depends(oauth2_scheme)):**
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

The additional header WWW-Authenticate with value Bearer we are returning here is also part of the spec.

Any HTTP (error) status code 401 "UNAUTHORIZED" is supposed to also return a WWW-Authenticate header.

This is how you implement an insecure authentication, now we will implement a secure version for our Todo API, get ready to code.

Adding Authentication Todo API

We will use the password flow with hashing for our Todo API and use JWT token instead of insecure usernames.

(You can now code along)

JWT

JWT means "JSON Web Tokens". It's a standard to codify a JSON object in a long dense string without spaces.

Example: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

It is not encrypted, so, anyone could recover the information from the contents.

But it's signed. So, when you receive a token that you emitted, you can verify that you actually emitted it.

That way, you can create a token with an expiration of, let's say, 1 week. And then when the user comes back the next day with the token, you know that the user is still logged in to your system.

After a week, the token will be expired and the user will not be authorized and will have to sign in again to get a new token. And if the user (or a third party) tried to modify the token to change the expiration, you would be able to discover it, because the signatures would not match.

Installpython-jose to generate and verify the JWT tokens in Python:

$ pip install python-jose[cryptography]

Password Hashing

Install PassLib to handle password hashes.

It supports many secure hashing algorithms and utilities to work with them. The recommended algorithm is "Bcrypt".

$ pip install passlib[bcrypt]

Let's create a file security.py inside the app directory:

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str):
    return pwd_context.hash(password)

We are creating a PassLib "context" that we will use to hash and verify passwords.

Handle JWT tokens

As told before, we are required to send an access token and token type as a response from the /token endpoint. That's why create a file [token.py](http://token.py) inside the schema folder to store the following Pydantic models. Token will be later used in the /token endpoint and TokenData will be used to store the decoded token payload.

from typing import Optional
from pydantic import BaseModel

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

Note: Add the following line to __init__.py inside schemas/ to simplify imports: from .token import Token, TokenData

Now, create a random secret key to sign the JWT tokens, execute the following command in your terminal:

$ openssl rand -hex 32

You will get a random key in the output like this:

939aae726cc09e3338ea836b90b0b0a4dffce9e8ed069ac343754c236244e3d2

Save this key in an environment variable(don't use this one). You can use a .env file to store it so that you don't have to manually set them every time you start a new shell.

create .env file in app folder:

SECRET_JWT_KEY=939aae726cc09e3338ea836b90b0b0a4dffce9e8ed069ac343754c236244e3d2

Install the python-dotenv package to import variables from the .env file into the environment: pip install python-dotenv.

Note: if you are using git, make sure to add .env in the .gitignore file as it contains sensitive information.

Let's add functions to handle tokens and authentication in the security.py file:

import os
from datetime import datetime, timedelta
from typing import Optional

from dotenv import load_dotenv
from fastapi import HTTPException, status
from jose import JWTError, jwt

from app import schemas

load_dotenv()  # load the variables from .env

SECRET_JWT_TOKEN = str(os.environ["SECRET_JWT_KEY"])
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)

    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_JWT_TOKEN, algorithm=ALGORITHM)

    return encoded_jwt

def decode_access_token(token: str):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearers"},
    )
    try:
        payload = jwt.decode(token, SECRET_JWT_TOKEN, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = schemas.TokenData(username=username)
    except JWTError:
        raise credentials_exception

    return token_data.username

Here, we are setting the variables for our secret key, algorithm, and expiration for the JWT token.

We are also defining a Pydantic model to use as a response model in the /token endpoint. TokenData model will store the data from the decoded JWT.

Then we finally created a function to generate an access token.

Note: We are using the crud module we haven't created yet, we will implement the CRUD functionality in the database section.

Creating auth dependencies

Create a file [dependency.py](http://dependency.py) in the app directory:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, oauth2
from sqlalchemy.orm import Session

from app import crud
from app.security import decode_access_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
    username = decode_access_token(token)
    user = crud.get_user_by_username(db, username)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

    return user

Token Route

Finally, let's create the endpoint for /token in the [main.py](http://main.py) inside app/:

(The get_db dependency will be created in the database section)

from datetime import timedelta

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from app import crud, schemas, security
from app.dependency import get_db

app = FastAPI()

@app.get("/")
async def root():
    return {"app": "Todo API"}

ACCESS_TOKEN_EXPIRE_MINUTES = 30

@app.post("/token", response_model=schemas.Token)
async def login(
    db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()
):
    user = crud.authenticate_user(db, form_data.username, form_data.password)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = security.create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

Here authenticate_user will be implemented in the next section of CRUD operations. It just checks if the user exists and has the correct password or not and returns a Boolean value.

JWT "subject" sub: The JWT specification says that there's a key sub, with the subject of the token. It's optional to use it, but that's where you would put the user's identification, so we are using it here.

The "sub" key should have a unique identifier across the entire application, and it should be a string.

Adding endpoints in the User route

Create a file [users.py](http://users.py) in the routers folder:

(User schema will be created in the database section)

from app import schemas
from fastapi import APIRouter
from fastapi.param_functions import Depends
from app.dependency import get_current_user

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/me", response_model=schemas.User)
async def read_users_me(current_user: schemas.users.User = Depends(get_current_user)):
    return current_user

To organize different path operations, we can use the APIRouter. It's similar to the FastAPI class and supports all the same options like dependencies, tags.

The prefix will be used for all the paths, so /me/ will become /users/me/.

We can also add a list of dependencies that will be added to all the path operations in the router and will be executed/solved for each request made to them.

Let's add this router to the [main.py](http://main.py) file:

from app.routers import users

app = FastAPI()

app.include_router(users.router)

Database

We have added the code authentication, but we need a database to store the users and perform operations on it before we can run it.

We will use the SQLite database for this article as Python has integrated support for it, although FastAPI doesn't require you to use a SQL (relational) database. You can use any relational or no SQL database with FastAPI.

(Later, for your production application, you might want to use a database server like PostgreSQL.)

ORMs

ORM stands for object-relational mapping, an ORM has tools to convert between objects in code and database tables(relations).

With an ORM, we can represent a table in a SQL database using a class. Each attribute of the class represents a column, with a name and a type.

Each instance of the class will represent a row in the table.

• SQL table —> ORM class

• SQL columns —> ORM class attribute

• SQL rows —> ORM class instances

In this article, we will use SQLAlchemy ORM.

Install SQLAlchemy by running:

$ pip install SQLAlchemy

Configuring SQLAlchemy

Let's start by configuring SQLAlchemy, create a file [database.py](http://database.py) inside the db folder.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./app/todo_app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

We first imported the SQLAlchemy tools and then created a database URL to connect to an SQLite database.

If you want to use a PostgreSQL database, change the URL to:

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

SQLAlchemyengine

Then we created an engine that is the starting point for any SQLAlchemy application, we will use it later.

Note:

The argument:

connect_args={"check_same_thread": False}

is needed only for SQLite. It's not needed for other databases.

SessionLocal class

Each instance of the SessionLocal class will be a database session. The class itself is not a database session yet.

But once we create an instance of the SessionLocal class, this instance will be the actual database session.

Base class

The function declarative_base() returns a class that is used to create each of the database models or classes.

Creating the database models for User

The classes used to interact with the databases are called models.

This is different than the previously mentioned Pydantic models which are used for data validation. To avoid confusion we will keep Pydantic models in the schemas folder and database models in the models' folder.

Let's create a file [users.py](http://users.py) inside the models' directory.

from app.db.database import Base
from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, nullable=False, index=True)
    hashed_password = Column(String, nullable=False)

We are using the Base class from [database.py](http://database.py) to create the User model.

The __tablename__ attribute tells SQLAlchemy the name of the table to use in the database for each of these models.

Creating model attributes/columns

Then we created the model(class) attributes: id, username, hashed_password.

Each attribute represents a column in the users' table, and the value is given using the Column from SQLAlchemy.

The type of attribute is the first argument in the Column.

Note: Add the following line to __init__.py inside models/ to simplify imports: from .users import User

Creating the Pydantic models for User

Create a file [users.py](http://users.py) inside the schemas directory:

from pydantic import BaseModel

class UserBase(BaseModel):
    username: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int

    class Config:
        orm_mode = True

We have created three pydantic models, for the creation (UserCreate), reading(User), and one base(UserBase) that has common attributes.

Password is only required when creating a user and should not be sent in a response.

The reading model User contains an id attribute as we will already know its ID after creation from the database.

The Config class is used to provide configurations to Pydantic. Pydantic's orm_mode will tell the Pydantic model to read the data even if it is not a dict, but an ORM model (or any other arbitrary object with attributes).

This way, instead of only trying to get the id value from a dict, as in:

id = data["id"]

it will also try to get it from an attribute, as in:

id = data.id

And with this, the Pydantic model is compatible with ORMs, and you can just declare it in the response_model argument in your path operations.

Note: Add the following line to __init__.py inside schemas/ to simplify imports: from .users import User, UserCreate

Creating CRUD for User

Create a file [users.py](http://users.py) inside the crud folder to define the functions to interact with the database's users' table.

CRUD: Create, Read, Update, and Delete.

For users, we will only implement create and read. Later, for todos, we will implement all four.

Read Users

from app import models, schemas
from app.security import get_password_hash, verify_password
from sqlalchemy.orm import Session

def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_username(db: Session, username: str):
    return db.query(models.User).filter(models.User.username == username).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()

We imported Session from sqlalchemy.orm, this will allow us to declare the type of the db parameters and have better type checks and completion in your functions.

Create Users

Add the following function to the same file:

  def create_user(db: Session, user: schemas.UserCreate):
    hashed_password = get_password_hash(user.password)
    db_user = models.users.User(username=user.username, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

• Create an SQLAlchemy model instance with your data.

• add that instance object to your database session.

• commit the changes to the database (so that they are saved).

• refresh your instance (so that it contains any new data from the database, like the generated ID).

Authenticate Users

Also, adds the following function:

def authenticate_user(db: Session, username: str, password: str):
    user = get_user_by_username(db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

Note: Add the following line to __init__.py inside crud/ to simplify imports: from .users import get_user, get_user_by_username, get_users, create_user, authenticate_user.

Initializing the database and setting migrations with Alembic

We have written then needed SQLAlchemy code for the users' table, but we need to convert it into a real database, for this, we use a migration tool.

A "migration" is the set of steps needed whenever we change the structure of our SQLAlchemy models, add a new attribute, etc. to replicate those changes in the database, add a new column, a new table, etc.

We will use Alembic for this purpose.

We will cover the necessary parts of alembic in this article. You can learn more in the alembic docs.

Install alembic

$ pip install alembic

Create Migrations Environment

Run the following command inside the todo directory:

$ alembic init alembic

This will create a migration environment for us.

A migration environment is a directory of scripts:

todo/
    alembic/
        env.py
        README
        script.py.mako
        versions/

• alembic/: home of the migration environment.

• env.py: a Python script that is run whenever the alembic migration tool is invoked. It configures and generates an SQLAlchemy engine.

• script.py.mako: a Mako template file to generate new migration scripts inside versions.

• versions/: This directory holds the individual version scripts.

Configurealembic.ini

Alembic created a file alembic.ini in the todo folder. This is a file that the alembic script looks for when invoked.

Edit the following line:

sqlalchemy.url = driver://user:pass@localhost/dbname

to:

sqlalchemy.url = sqlite:///./app/todo_app.db

Creating migrations scripts

We can create migration scripts either manually or automatically, here we will use the automatic approach and manually edit the scripts if required.

Alembic can view the status of the database and compare it against the table metadata in the application, generating the “obvious” migrations based on a comparison.

Create a file base.py to contain all the models for alembic to see:

# Import all the models, so that Base has them before being
# imported by Alembic
from app.db.database import Base
from app.models.users import User

Open the [env.py](http://env.py) file inside the alembic folder and make the following changes to target_metadata = None:

from app.db.base import Base

target_metadata = Base.metadata

Also, make sure the run_migrations_online() function is same as follows:

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()

Now, run the alembic revision command with the --autogenerate option to create the script.

$ alembic revision --autogenerate -m "Add users table"

• The -m flag adds a message to the script.

A new script is created inside the versions/ folder, we can alter it as needed but it's not required in this case.

Running the migration script

Running the script will reflect the newly made changes in the actual database.

Run the following command to get to the "the most recent" version of database - head

$ alembic upgrade head

Yay! The database is initialized and we have our table live in the database.

Interact with the database directly

Before interacting with the database through FastAPI, let's do it directly.

Download the SQLiteBrowser and open the todo_app.db file with it:

$ sqlitebrowser app/todo_app.db # Or use the GUI in windows

Using FastAPI to interact with the database

Creating database dependency

Open the [dependency.py](http://dependency.py) in the app folder and create the following dependency:

from app.database import SessionLocal

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

We are using the SessionLocal class created in the [databases.py](http://databases.py) file.

We need to have an independent database session/connection (SessionLocal) per request, use the same session through all the requests, and then close it after the request is finished.

And then a new session will be created for the next request.

Creating Path Operations

Now, we will use the dependency created above in the path operations function:

Add the following functions in the [users.py](http://users.py) file inside routers, see how we are using get_current_user dependency to make authentication required for an operation.

from typing import List
from app import crud, schemas
from app.dependency import get_current_user, get_db
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=schemas.User)
async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_username(db, username=user.username)
    if db_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="Username already exists"
        )
    return crud.create_user(db, user)

@router.get(
    "/", response_model=List[schemas.User], dependencies=[Depends(get_current_user)]
)
async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

@router.get("/me", response_model=schemas.User)
async def read_users_me(current_user: schemas.User = Depends(get_current_user)):
    return current_user

@router.get(
    "/{user_id}", response_model=schemas.User, dependencies=[Depends(get_current_user)]
)
async def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id)
    if db_user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
        )
    return db_user

main.py

Add the following in main.py, It initializes the tables in the database:

from app.db.database import engine
from app.db import base

base.Base.metadata.create_all(bind=engine)

app = FastAPI()

Running the app with Auth and Users

Let's check our progress till now by running the app:

$ uvicorn app.main:app --reload

Try creating users, signing in with the credentials of the new user-created, and then retrieving the multiple users or a single one with all the path operations.

Creating Todo Items

Now, we will create CRUD operations for Todo Items

Create Todo Schemas

Let's start by creating Pydantic models for a to-do item.

Create a file [todo.py](http://todos.py) in the schemas/ folder:

from typing import Optional

from pydantic import BaseModel

class TodoBase(BaseModel):
    title: str
    description: Optional[str] = None
    done: bool = False

class TodoCreate(TodoBase):
    pass

class Todo(TodoBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

Also, add the following line to __init__.py of schemas/: from .todo import Todo, TodoCreate

Create Todo SQLAlchemy Models

Create a file [todo.py](http://todo.py) inside models/:

from app.db.database import Base
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

class Todo(Base):
    __tablename__ = "todos"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    description = Column(String)
    done = Column(Boolean, default=False)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="todos")

You are familiar with most of the things happening here, only the relationship thing is new.

A user and a todo are related to each other by the "owns" relationship such that a user owns a todo, we express this in a database using a foreign key that relates two entities together.

As one user can have multiple todo items but a todo item can have only one user associated with it, we placed this foreign key inside the todo item table.

The real magic lies in how easily we can access which user owns a todo item and all the todo items created by a user.

When accessing the attribute owner in a todo item, it will contain a User model from the users' table. SQLAlchemy will use the owner_id attribute/column with its foreign key to know which record to get from the users' table.

You also have to reflect this relationship inside the User model, so change it as follows:

from app.db.database import Base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, nullable=False, index=True)
    hashed_password = Column(String, nullable=False)

    todos = relationship("Todo", back_populates="owner")

Now, accessing user.todos will give you a list of all the todo items created by that user.

Add the following line to __init__.py of models/: from .todo import Todo

Finally, update [base.py](http://base.py) in database/ to have the Todo model:

from app.db.database import Base
from app.models.users import User
from app.models.todo import Todo

Let's create a migration script and run the migrations to reflect the changes in the database:

$ alembic revision --autogenerate -m "Add todos table"

$ alembic upgrade head

You can check the database using sqlitebrowser, a new table would have been created for todo items.

Creating CRUD operations

We have to do the following operations on a todo item:

1. Create a new to-do item.

2. View all to-do items.

3. Mark a to-do item as completed.

4. Delete an existing to-do item.

Create a file [todo.py](http://todo.py) inside crud/.

1. Create Todo Item

    from app import models, schemas
    from sqlalchemy.orm import Session
   
    def create_todo(db: Session, todo: schemas.TodoCreate, user_id: int):
        db_todo = models.Todo(**todo.dict(), owner_id=user_id)
        db.add(db_todo)
        db.commit()
        db.refresh(db_todo)
        return db_todo
   

2. View all to-do items of current user

    def get_todos(db: Session, user_id: int, skip: int = 0, limit: int = 100):
        return db.query(models.Todo).filter(models.Todo.owner_id == user_id).offset(skip).limit(limit).all()
   

3. Mark a todo item as completed

    def update_todo_complete(db: Session, todo_id: int, complete: bool, user_id: int):
        db_todo = db.query(models.Todo).get(todo_id)
        if not db_todo:
            return None
        if db_todo.owner_id != user_id:
            raise Exception("Not authorized")
        db_todo.done = complete
        db.commit()
        db.refresh(db_todo)
        return db_todo
   

4. Delete an existing to-do item

    def delete_todo(
        db: Session,
        todo_id: int,
        user_id: int,
    ):
        db_todo = db.query(models.Todo).get(todo_id)
   
        if not db_todo:
            raise KeyError("Todo Not found")
   
        if db_todo.owner_id != user_id:
            raise Exception("Not authorized")
   
        db.delete(db_todo)
        db.commit()
   

These are the required four functions, let's add them to __init__.py of crud/: from .todo import create_todo, get_todos, update_todo_complete, delete_todo

Creating Router for To-do

It's time to create path operations for to-do items, create a file [todo.py](http://todo.py) inside routers/

from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from app import crud, schemas
from app.dependency import get_db, get_current_user
from sqlalchemy.orm import Session

router = APIRouter(prefix="/todos", tags=["todos"])

1. Path for creating to-do

    @router.post("/", response_model=schemas.Todo)
    async def create_todo(
        todo: schemas.TodoCreate,
        db: Session = Depends(get_db),
        current_user: schemas.User = Depends(get_current_user),
    ):
        todo = crud.create_todo(
            db,
            todo,
            current_user.id,
        )
        return todo
   

2. Path for viewing all todos

    @router.get("/", response_model=List[schemas.Todo])
    async def read_all_todos(
        skip: int = 0,
        limit: int = 100,
        db: Session = Depends(get_db),
        current_user: schemas.User = Depends(get_current_user),
    ):
        todos = crud.get_todos(db, current_user.id, skip, limit)
        return todos
   

3. Path for updating a to-do as complete/incomplete

    @router.put("/", response_model=schemas.Todo)
    async def update_todo_complete(
        todo_id: int,
        complete: bool,
        db: Session = Depends(get_db),
        current_user: schemas.User = Depends(get_current_user),
    ):
        try:
            todo = crud.update_todo_complete(
                db,
                todo_id,
                complete,
                current_user.id,
            )
        except Exception as e:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
   
        if todo is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found"
            )
        return todo
   

4. Path for deleting a to-do item

    @router.delete("/{todo_id}")
    async def delete_todo(
        todo_id: int,
        db: Session = Depends(get_db),
        current_user: schemas.User = Depends(get_current_user),
    ):
        try:
            crud.delete_todo(db, todo_id, current_user.id)
        except KeyError as e:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
        except Exception as e:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
   
        return "Todo Deleted"
   

Let's add the router to main.py

from app.routers import users, todo

base.Base.metadata.create_all(bind=engine)

app = FastAPI()

app.include_router(users.router)
app.include_router(todo.router)

Let's run the app:

$ uvicorn app.main:app --reload

Testing

Yay! We have created an API that does what we want. But does it?

Let's write some tests for the CRUD operations and API routes to validate our code.

What is Testing?

Testing is the act of making sure everything runs as we expect it to.

Let's see a simple example:

def square(x):
    return x * x

We can test the above function by using an assert statement:

assert square(10) == 100

If the square function we wrote is working properly the assert would do nothing, but if the square function is defined incorrectly then an AssertionError is raised.

Testing in FastAPI

For testing our FastAPI app we will use the pytest library to make things easy.

Install the pytest library by executing the following command:

$ pip install -U pytest

Create a new folder tests in the app directory.

Now, create a file named [conftest.py](http://conftest.py) inside tests. This file will be used to create a temporary database and fixtures.

What are fixtures?

Fixtures are functions decorated with a @pytest.fixture decorator.

When pytest goes to run a test, it looks at the parameters in that test function’s signature and then searches for fixtures that have the same names as those parameters. Once pytest finds them, it runs those fixtures, captures what they returned (if anything), and passes those objects into the test function as arguments.

• pytest docs

One use case of a fixture is to clear a database after each test and create a new one before each test.

Let's first create a temporary database for running tests so that we don't accidentally change our actual database.

Add the following in the [conftest.py](http://conftest.py) file:

import pytest
from app.dependency import get_db
from sqlalchemy.orm import Session, sessionmaker
from app.db.database import Base
from app.main import app

SQLALCHEMY_DATABASE_URL = "sqlite:///./app/test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="session", autouse=True)
def db_setup():
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

@pytest.fixture(scope="session")
def db() -> Generator:
    yield from override_get_db()

app.dependency_overrides[get_db] = override_get_db

A lot is going on here, let's break it down:

• First, we created a new database using the same SQLAlchemy function we used before but this time we used a different SQLALCHEMY_DATABASE_URL. This will create a new database file test.db instead of using the previous todo_app.db.

• Then, we created a fixture:

    @pytest.fixture(scope="session", autouse=True)
    def db_setup():
        Base.metadata.create_all(bind=engine)
        yield
        Base.metadata.drop_all(bind=engine)
  

  • The scope of a fixture defines when it will be destroyed. Here are other possible values for scope:

    Fixtures are created when first requested by a test, and are destroyed based on their scope:

    • function: the default scope, the fixture is destroyed at the end of the test.

    • class: the fixture is destroyed during the teardown of the last test in the class.

    • module: the fixture is destroyed during the teardown of the last test in the module.

    • package: the fixture is destroyed during the teardown of the last test in the package.

    • session: the fixture is destroyed at the end of the test session.

  • The autouse=True means that this fixture will automatically be used for every test without explicitly mentioning the fixture name, it makes sense because we will use a database for all tests.

  • Then in the function, the part above yield will be executed before running any test, and the part after will run after all the tests are completed.

    Before yield, we are creating a new database, and after yield, we are clearing that database.

• Then we created a new dependency that will use the temporary database. We override the get_db dependency with this one so any function in our app that uses the get_db will now use override_get_db, as a result, using the temporary database. This allowed us to use a new database without changing any code in our app.

• We also created another fixture db to access the database directly without the dependencies. This will help write tests for CRUD operations.

These things will be more clear when we use them in our tests.

Let's add three more fixtures to the conftest.py:

from typing import Dict, Generator
from app import models
from fastapi.testclient import TestClient
from . import utils    # utils.py is created next

@pytest.fixture(scope="module")
def client() -> Generator:
    with TestClient(app) as c:
        yield c

@pytest.fixture(scope="function")
def user(db: Session) -> models.User:
    return utils.create_random_user(db)

@pytest.fixture(scope="module")
def user_token_headers(client: TestClient, db: Session) -> Dict[str, str]:
    return utils.authentication_token(client=client, db=db)

Let's go through each fixture:

• The client fixture: We will use this to make API calls from our code like the get and post request. It is based on Python's request library.

• The user fixture: It will create random users to test the todo CRUD operations.

• the user_token_headers fixture: It is used to provide bearer token while accessing endpoints which needs authentication.

A few fixtures above uses functions from the [utils.py](http://utils.py) file, let's define it in the same tests folder:

import random
import string
from typing import Dict
from app import crud, models, schemas

from sqlalchemy.orm import Session
from fastapi.testclient import TestClient

def random_lower_string() -> str:
    return "".join(random.choices(string.ascii_lowercase, k=random.randint(1, 32)))

def create_random_user(db: Session) -> models.User:
    username = random_lower_string()
    password = random_lower_string()
    user_in = schemas.UserCreate(username=username, password=password)
    user = crud.create_user(db, user=user_in)
    return user

def user_authentication_headers(
    *, client: TestClient, username: str, password: str
) -> Dict[str, str]:
    data = {"username": username, "password": password}
    r = client.post("/token", data=data)
    response = r.json()
    auth_token = response["access_token"]
    headers = {"Authorization": f"Bearer {auth_token}"}
    return headers

def authentication_token(*, client: TestClient, db: Session) -> Dict[str, str]:
    username = random_lower_string()
    password = random_lower_string()
    crud.create_user(db, user=schemas.UserCreate(username=username, password=password))
    return user_authentication_headers(
        client=client, username=username, password=password
    )

Here's what each function is doing:

• random_lower_string: Creates a random string of max length 32 to be used in the test to create users and todo items.

• create_random_user: Creates and returns a random user.

• user_authentication_headers: Returns the authorization header to be used to authenticate API requests made through tests.

• authentication_token: Creates a new user and returns its authorization header.

Tests for CRUD operations

We are now done with the setup required to test our code, now we will test the CRUD operations for users and todos.

Create folder crud inside tests.

Create a file called test_user.py inside crud. Remember to prepend the file name by test_, this is how pytest will find test files to execute.

Let's define the test functions for each of the crud operation on user one by one:

Import the necessary modules

from app import crud
from app.schemas import UserCreate
from app.tests import utils
from sqlalchemy.orm import Session
from app.tests import utils
from app import models

Test creating a user

def test_create_user(db: Session):
    username = utils.random_lower_string()
    password = utils.random_lower_string()
    user_in = UserCreate(username=username, password=password)
    user = crud.create_user(db, user=user_in)
    assert user.username == username
    assert hasattr(user, "hashed_password")

This test function corresponds to the create_user function of crud/users.py. We are creating a user and then making sure the created user has the same username as the one used to create that user and has the hashed_password attribute.

Remember each test function's name must start with test_ and a test function's name should be descriptive.

Test authenticating a user

def test_authenticate_user(db: Session):
    username = utils.random_lower_string()
    password = utils.random_lower_string()
    user_in = UserCreate(username=username, password=password)
    user = crud.create_user(db, user=user_in)
    authenticated_user = crud.users.authenticate_user(
        db, username=username, password=password
    )
    assert authenticated_user
    assert authenticated_user.username == user.username

This function corresponds to the authenticate_user function of the crud/users.py. We are making sure an existing user with the correct credentials is authenticated.

Test not authenticating a user

def test_not_authenticate_user(db: Session):
    username = utils.random_lower_string()
    password = utils.random_lower_string()
    user_in = UserCreate(username=username, password=password)
    user = crud.create_user(db, user=user_in)
    assert not crud.authenticate_user(db, username=username, password="wrong")
    assert not crud.authenticate_user(db, username="wrong", password=password)
    assert not crud.authenticate_user(db, username="wrong", password="wrong")

This function also corresponds to the authenticate_user, but this time it is checking a different path that is when a user does not exists or enter incorrect credentials.

Test getting a user by id

def test_get_user_by_id(db: Session, user: models.User):
    user_out = crud.get_user(db, user_id=user.id)
    assert user_out
    assert user_out.username == user.username
    assert user_out.hashed_password == user.hashed_password

We are testing crud.get_user here, first we created a new user and called get_user with its id and checked if we are getting the correct user in output or not.

We also used the userfixture in the arguments**,** so the user parameter will be provided with a random user by pytest.

Test getting a user by username

def test_get_user_by_username(db: Session, user: models.User):
    user_out = crud.get_user_by_username(db, username=user.username)
    assert user_out
    assert user_out.username == user.username
    assert user_out.hashed_password == user.hashed_password

This one is similar to the previous one.

Testing getting all users

def test_get_users(db: Session, user: models.User):
    user_out = crud.get_users(db)
    print(user_out)
    assert user_out
    assert user_out[-1].username == user.username
    assert user_out[-1].hashed_password == user.hashed_password

Here, we are creating a new user using the fixture and getting all the users, and finally checking that the new user is the last one in the list or not.

That's it for test for user's crud operation.

Try running them with the following command in the todo directory:

$ python -m pytest app/tests/crud

You will get an output like this:

Each green dot represents a single test function and means that the test passed, an 'F' means that the test failed.

Here are the test functions for todo's crud operations(crud/todo.py), you will be able to understand them now:

Create a file test_todo.py in the tests/crud folder and the following tests:

import pytest
from app import crud, models, schemas
from app.tests import utils
from sqlalchemy.orm import Session

def test_create_todo(db: Session, user: models.User) -> None:
    """Test creating a todo"""
    todo_in = schemas.TodoCreate(
        title=utils.random_lower_string(),
        description=utils.random_lower_string(),
        done=False,
    )
    todo_out = crud.create_todo(db=db, todo=todo_in, user_id=user.id)
    assert todo_out.title == todo_in.title
    assert todo_out.owner_id == user.id

def test_get_todos(db: Session, user: models.User) -> None:
    """Test getting all todos"""
    todo_in_1 = schemas.TodoCreate(title=utils.random_lower_string())
    crud.create_todo(db=db, todo=todo_in_1, user_id=user.id)
    todo_in_2 = schemas.TodoCreate(title=utils.random_lower_string())
    crud.create_todo(db=db, todo=todo_in_2, user_id=user.id)
    todos = crud.get_todos(db=db, user_id=user.id)
    assert len(todos) == 2
    assert todos[0].title == todo_in_1.title
    assert todos[1].title == todo_in_2.title

def test_update_todo_complete(db: Session, user: models.User) -> None:
    """Test updating a todo's complete status"""
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    todo = crud.create_todo(db=db, todo=todo_in, user_id=user.id)
    todo_out = crud.update_todo_complete(
        db=db, todo_id=todo.id, complete=True, user_id=user.id
    )
    assert todo_out.done is True

def test_update_todo_complete_when_todo_does_not_exist(
    db: Session, user: models.User
) -> None:
    """Test updating a todo's complete status when todo doesn't exist"""
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    todo = crud.create_todo(db=db, todo=todo_in, user_id=user.id)
    todo_out = crud.update_todo_complete(
        db=db, todo_id=todo.id + 1, complete=True, user_id=user.id
    )
    assert todo_out is None

def test_update_todo_complete_when_todo_is_not_owned(
    db: Session, user: models.User
) -> None:
    """Test updating a todo's complete status when todo is not owned"""
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    todo = crud.create_todo(db=db, todo=todo_in, user_id=user.id)
    with pytest.raises(Exception) as e:
        todo_out = crud.update_todo_complete(
            db=db, todo_id=todo.id, complete=True, user_id=user.id + 1
        )

def test_delete_todo(db: Session, user: models.User) -> None:
    """Test deleting a todo"""
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    todo = crud.create_todo(db=db, todo=todo_in, user_id=user.id)
    crud.delete_todo(db=db, todo_id=todo.id, user_id=user.id)
    todos = crud.get_todos(db=db, user_id=user.id)
    assert todo not in todos

def test_delete_todo_when_todo_does_not_exist(db: Session, user: models.User) -> None:
    """Test deleting a todo when todo doesn't exist"""
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    todo = crud.create_todo(db=db, todo=todo_in, user_id=user.id)
    with pytest.raises(KeyError) as e:
        crud.delete_todo(db=db, todo_id=todo.id + 1, user_id=user.id)
    todos = crud.get_todos(db=db, user_id=user.id)
    assert todo in todos

def test_delete_todo_when_todo_is_not_owned(db: Session, user: models.User) -> None:
    """Test deleting a todo when todo is not owned"""
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    todo = crud.create_todo(db=db, todo=todo_in, user_id=user.id)
    with pytest.raises(Exception) as e:
        crud.delete_todo(db=db, todo_id=todo.id, user_id=user.id + 1)
    todos = crud.get_todos(db=db, user_id=user.id)
    assert todo in todos

Try running them.

Now we will write and understand test functions for the API.

Tests for API endpoints

Create folder api inside tests and create a file test_users.py.

Import necessary modules intest_users.py

from typing import Dict

from app import models, schemas
from fastapi.testclient import TestClient
from app.tests import utils

Test Create User

def test_create_user(client: TestClient):
    user = schemas.UserCreate(username="test", password="test")
    response = client.post("/users/", json=user.dict())
    assert 200 <= response.status_code < 300
    created_user = response.json()
    assert created_user["username"] == user.username

Here we are testing the /users/ endpoint when accessed with a post request. We are using the client fixture to make the request, the first parameter to [client.post](http://client.post) defines the path and the json parameter includes the json data needed to create a user.

Then we are asserting a successful response and creation of a correct user.

Here are a few ways to send data in a request using the client

• To pass a path or query parameter, add it to the URL itself.

• To pass a JSON body, pass a Python object (e.g. a dict) to the parameter json.

• If you need to send Form Data instead of JSON, use the data parameter instead.

• To pass headers, use a dict in the headers parameter.

• For cookies, a dict in the cookies parameter.

Remaining test functions for User's endpoint

Similarly, we can create tests for other endpoints:

def test_create_user_when_exist(client: TestClient):
    user = schemas.UserCreate(username="existing", password="existing")
    response = client.post("/users/", json=user.dict())
    assert 200 <= response.status_code < 300
    response = client.post("/users/", json=user.dict())
    assert 400 <= response.status_code < 500

def test_read_users(client: TestClient, user_token_headers: Dict[str, str]):
    user1 = schemas.UserCreate(
        username=utils.random_lower_string(), password=utils.random_lower_string()
    )
    response = client.post("/users/", json=user1.dict())
    assert 200 <= response.status_code < 300

    user2 = schemas.UserCreate(
        username=utils.random_lower_string(), password=utils.random_lower_string()
    )
    response = client.post("/users/", json=user2.dict())
    assert 200 <= response.status_code < 300

    response = client.get("/users/", headers=user_token_headers)
    assert 200 <= response.status_code < 300
    users = response.json()
    assert len(users) > 0
    assert users[-1]["username"] == user2.username

def test_read_user_me(client: TestClient, user_token_headers: Dict[str, str]):
    response = client.get("/users/me", headers=user_token_headers)
    assert 200 <= response.status_code < 300
    current_user = response.json()
    assert current_user
    assert "username" in current_user

def test_read_user(
    client: TestClient, user_token_headers: Dict[str, str], user: models.User
):
    user_id = user.id
    response = client.get(f"/users/{user_id}", headers=user_token_headers)
    assert 200 <= response.status_code < 300
    existing_user = response.json()
    assert existing_user
    assert existing_user["username"] == user.username

def test_read_user_when_not_found(
    client: TestClient, user_token_headers: Dict[str, str], user: models.User
):
    response = client.get(f"/users/{user.id + 1}", headers=user_token_headers)
    assert response.status_code == 404

Also, create a file test_todos.py in api and add the following tests:

from typing import Dict
from app import schemas
from fastapi.testclient import TestClient
from app.tests import utils

def test_create_todo(client: TestClient, user_token_headers: Dict[str, str]):
    todo_in = schemas.TodoCreate(
        title=utils.random_lower_string(), description=utils.random_lower_string()
    )
    response = client.post("/todos/", json=todo_in.dict(), headers=user_token_headers)

    assert 200 <= response.status_code < 300

    todo = response.json()
    assert todo["title"] == todo_in.title

def test_read_all_todos(client: TestClient, user_token_headers: Dict[str, str]):
    todo1 = schemas.TodoCreate(
        title=utils.random_lower_string(), description=utils.random_lower_string()
    )
    response = client.post("/todos/", json=todo1.dict(), headers=user_token_headers)
    assert 200 <= response.status_code < 300

    todo2 = schemas.TodoCreate(
        title=utils.random_lower_string(), description=utils.random_lower_string()
    )
    response = client.post("/todos/", json=todo2.dict(), headers=user_token_headers)
    assert 200 <= response.status_code < 300

    response = client.get("/todos/", headers=user_token_headers)
    assert 200 <= response.status_code < 300

    todos = response.json()
    assert len(todos) > 0
    assert todos[-1]["title"] == todo2.title

def test_update_todo_complete(client: TestClient, user_token_headers: Dict[str, str]):
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    response = client.post("/todos/", json=todo_in.dict(), headers=user_token_headers)
    assert 200 <= response.status_code < 300

    todo = response.json()
    assert todo["title"] == todo_in.title

    response = client.put(
        f"/todos/?todo_id={todo['id']}&complete=true", headers=user_token_headers
    )
    assert 200 <= response.status_code < 300

    todo = response.json()
    assert todo["title"] == todo_in.title
    assert todo["done"] is True

def test_update_todo_complete_when_not_found(
    client: TestClient, user_token_headers: Dict[str, str]
):
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    response = client.post("/todos/", json=todo_in.dict(), headers=user_token_headers)
    assert 200 <= response.status_code < 300

    todo = response.json()
    assert todo["title"] == todo_in.title

    response = client.put(
        f"/todos/?todo_id={todo['id'] + 1}&complete=true", headers=user_token_headers
    )
    assert response.status_code == 404

def test_delete_todo(client: TestClient, user_token_headers: Dict[str, str]):
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    response = client.post("/todos/", json=todo_in.dict(), headers=user_token_headers)
    assert 200 <= response.status_code < 300

    todo = response.json()
    assert todo["title"] == todo_in.title

    response = client.delete(f"/todos/{todo['id']}", headers=user_token_headers)
    assert 200 <= response.status_code < 300

    response = client.get("/todos/", headers=user_token_headers)
    assert 200 <= response.status_code < 300

    todos = response.json()
    assert todo not in todos

def test_delete_todo_not_found(client: TestClient, user_token_headers: Dict[str, str]):
    todo_in = schemas.TodoCreate(title=utils.random_lower_string())
    response = client.post("/todos/", json=todo_in.dict(), headers=user_token_headers)
    assert 200 <= response.status_code < 300

    todo = response.json()
    assert todo["title"] == todo_in.title

    response = client.delete(f"/todos/{todo['id'] + 1}", headers=user_token_headers)
    assert response.status_code == 404

As an exercise, I encourage you to define a test function for the token/ endpoint from the [main.py](http://main.py) file.

With this we are done with testing, let's check the result.

Testing Result

Testing Coverage: This is a technique to determine how much application code we are testing.

Install the pytest-cov package which is a pytest plugin to measure test coverage.

$ install pip install pytest-cov

And run the following command in the todo directory:

$ python -m pytest app --cov=app --cov-report term-missing

We get the following output:

---------- coverage: platform linux, python 3.8.10-final-0 -----------
Name                          Stmts   Miss  Cover   Missing
-----------------------------------------------------------
...
app/crud/todo.py                 28      0   100%
app/crud/users.py                23      0   100%
...
app/routers/todo.py              32      4    88%   49-50, 69-70
app/routers/users.py             25      0   100%
...
-----------------------------------------------------------
TOTAL                           519     15    97%

=============== 26 passed, 6 warnings in 7.53s ================

We passed all 26 tests that we wrote and the test coverage is 97%.

Now try running the app with uvicorn and try out different endpoints yourself, try creating a user, signing in, creating and deleting todo items.

Woohoo!

Congratulations on making it this far.

We have created a Todo API with FastAPI and in the process learned a lot of things, I hope you have enjoyed it.

What's next?

I would recommend you to read the official docs of FastAPI and create your own Project.

FastAPI docs are one of the best I have seen with proper examples, even this article is highly inspired by the same and uses many examples of the docs.

You would now find it much easier to work your way through the docs and create structured projects with FastAPI.