FastAPI Introduction

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

# Installation and Startup#

First, install FastAPI

pip install "fastapi[all]"

The above installation also includes uvicorn, which can be used as a server to run the code.

To execute the code, use the following command

uvicorn main:app --reload

To start the server in the background

nohup uvicorn main:app --host 0.0.0.0 --port 8000 > /file/path/log.log 2>&1 &

Then execute

bg

This can switch a job that is paused in the background to continue executing in the background.

Finally, execute

disown

This is used to detach a job from the current terminal session so that it will not be affected by the terminal session and will continue running after exiting the terminal.

The meaning of the uvicorn main:app command is as follows:

main: the main.py file (a Python "module").
app: the object created in the main.py file via app = FastAPI(), which can also be another name.
--reload: allows the server to restart upon code updates. Use this option only during development.

At this point, access localhost:8000 to call it.

# Hello World#

## Create a main.py file#

from fastapi import FastAPI

app = FastAPI()

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

Code explanation

from fastapi import FastAPI

FastAPI is a Python class that provides all the functionality for your API.

app = FastAPI()

Here, the variable app will be an "instance" of the FastAPI class.

This instance will be the main interaction object for creating all APIs.

This app is also referenced by uvicorn in the startup command.

It can also be written as myapi = FastAPI()
The startup command would then change to uvicorn main --reload.

@app.get("/")

Tells FastAPI that the function below it is responsible for handling the following access request.

The request path is /

Called using the get operation.

You can also use @app.post(), @app.put(), @app.delete().

async def root():

Defines a Python function named root.

async indicates that this is an asynchronous function.

return {"message": "Hello World"}

Return content.

## Execute the file#

uvicorn main:app --reload

Execution successful.

# Path Parameters#

## Passing parameters in the path#

@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}

{item_id} indicates that this is a parameter that can be passed in.

At this point, requesting localhost:8000/items/1 will return

{ "item_id": "1"}

## Restricting parameter types#

At the same time, you can restrict the type of this parameter.

@app.get("/items/{item_id}")
async def read_item(item_id:int):
    return {"item_id": item_id}

Just add after the function parameter.

If a string is passed in at this point, a 422 error will be returned.

Types can be str, float, bool, etc.

Note!

Be aware of the order issue.

For example:

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/me")
async def read_user_me():
    return {"user_id": "the current user"}


@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

Now there are two interfaces, if /users/{user_id} is before /users/me,

Since path operations are executed in order,

When requesting /users/me, it will execute the /users/{user_id} interface and return the string "me".

So ensure that /users/me is before /users/{user_id}.

## Enum Type Default Values#

### Create an Enum Class#

from fastapi import FastAPI
# Import enum type
from enum import Enum

# Create an enum class
class ModelNameEnum(str, Enum):
      fire = "fire"
      peach = "peach"
      firepeach = "firepeach"

app = FastAPI()


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

@app.get("/items/{item_id}") 
async def read_item(item_id:str): 
    return {"item_id": item_id}

@app.get("/models/{model_name}")
async def get_model(model_name: ModelNameEnum):
    # Determine by the value of the enum class
    if model_name is ModelNameEnum.fire:
        return {"model_name": model_name, "message": "Fire Fire Fire"}
    if model_name.value == "peach":
        return {"model_name": model_name, "message": "Peach Peach Peach"}
    return {"model_name": model_name, "message": "FirePeach FirePeach FirePeach"}

Code explanation:

async def get_model(model_name: ModelNameEnum):

The defined parameter model_name is of the ModelNameEnum enum type.

if model_name is ModelNameEnum.fire:

Here, the is and == have the same effect, used to compare with enum members.

However, when using ==, you need to use the parameter .value to get the value.

model_name.value

### Test Request#

## File Path as a Parameter#

If the parameter to be passed is a path,

You can use

/files/{file_path:path}

As the request path.

@app.get("/files/{file_path:path}") 
async def read_file(file_path: str): 
    return {"file_path": file_path}

When needing to pass the path /home/test/text.txt,

The request parameter is localhost:8000/files//home/test/text.txt, note the two /.

# Query Parameters#

## Default Values#

When declaring other function parameters that are not path parameters, they will be automatically interpreted as "query string" parameters.

For example, the following code

from fastapi import FastAPI

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


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

In the function parameter

async def read_item(skip: int = 0, limit: int = 10):

Default values are defined, skip defaults to 0, limit defaults to 10.

When accessing http://127.0.0.1:8000/items/?skip=0&limit=10, the passed parameters are skip = 0, limit = 10.
When accessing http://127.0.0.1:8000/items/?skip=1, the passed parameters are skip = 1, limit = 10, because limit is not passed, it defaults to 10.
When accessing http://127.0.0.1:8000/items/?limit=1, the passed parameters are skip = 0, limit = 1, skip is not passed and defaults to 0.

## Optional Values#

Need to import Union.

from typing import Union

@app.get("/items/{item_id}")
async def read_itemUnion(item_id: str, q: Union[str, None] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

At this point, q is an optional parameter.

You can also add a bool type parameter.

@app.get("/items/{item_id}")
async def read_itemUnion(item_id: str, q: Union[str, None] = None, short: bool = False):
    item = {"item_id": item_id}
    if q:
        # Use q as a key-value pair for item, update can merge the new key-value pair.
        item.update({"q": q})
    if not short:
        item.update({"description": "This is an amazing item that has a long description"})
    return item

Using update({ "XX" : "XX" }), you can add this key-value pair to the existing key-value pairs.

Test access.

## Multiple Path and Query Parameters#

@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int, item_id: str, q: Union[str, None] = None, short: bool = False
):
    item = {"item_id": item_id, "owner_id": user_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update(
            {"description": "This is an amazing item that has a long description"}
        )
    return item

Can support requests with multiple parameters.

## Required Query Parameters#

If a parameter without a default value is specified among non-path parameters, then this parameter is required.

That is, it is not specified in @app.get(/XXX/{ XXX }), but specified in the function parameters.

And if Union[str, None] = None is not configured, then this parameter is required.

@app.get("/items/{item_id}") 
async def read_user_item(item_id: str, needy: str): 
    item = {"item_id": item_id, "needy": needy} 
    return item

For example, here the needy parameter is required.

# Request Body#

When you need to send data from the client (e.g., browser) to the API, you send it as a "request body".

The request body is the data sent by the client to the API. The response body is the data sent by the API to the client.

Your API will almost always send a response body. However, the client does not always need to send a request body.

## Import BaseModel#

from pydantic import BaseModel

You need to use the pydantic model to declare the request body.

## Create Data Model#

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

At this point, a structure type for the request body named Item has been declared.

{
    "name": "Foo",
    "description": "An optional description",
    "price": 45.2,
    "tax": 3.5
}

Next, add a post method to specify the parameter as the Item model type.

@app.post("/items/") 
async def create_item(item: Item): 
    return item

Test request.

## Using the Model#

Inside the function, you can directly access all properties of the model object.

However, you need to use model.dict() to get the model's dictionary for modification.

model.dict() extracts each property from the model.

@app.post("/items/") 
async def create_item(item: Item): 
    item_dict = item.dict() 
    if item.tax: 
        price_with_tax = item.price + item.tax 
        item_dict.update({"price_with_tax": price_with_tax}) 
    return item_dict

## Request Body with Path Parameter#

@app.put("/items/{item_id}")
async def create_item2(item_id: int, item: Item):
    return {"item_id": item_id, "item": item, **item.dict()}

You can pass both path parameters and request bodies simultaneously.

Note to add ** when returning item.dict().

It can be seen that returning item.dict() extracts each property from the model.

## Request Body, Path Parameter, and Query Parameter#

@app.put("/items/{item_id}")
async def create_item(item_id: int, item: Item, q: Union[str, None] = None):
    result = {"item_id": item_id, **item.dict()}
    if q:
        result.update({"q": q})
    return result

# Parameter Validation#

Query Parameter Validation#

First, you need to import Query.

from fastapi import FastAPI, Query

### Length Validation#

@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, min_length=2, max_length=5)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

You can add Query() after the parameter.

Since you must replace the default value None with Query(default=None), the first parameter of Query is also used to define the default value.

The above code indicates that the field q has a minimum length of 2 and a maximum length of 5.

### Regular Expression Validation#

@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, min_length=2, max_length=5, regex='^query$')):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

regex='^query$'

The above code specifies a regular expression, and the content of q must be query.

### Default Values#

Default values can also be specified.

default = 'query'

## Path Parameter Validation#

Need to import Path.

from fastapi import FastAPI, Query, Path

Add methods.

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    *,
    item_id: int = Path(title="The ID of the item to get", ge=0, le=1000),
    q: str,
    size: float = Query(gt=0, lt=10.5),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

@app.get("/items4/{item_id}")
async def read_items6(
    *,
    item_id: int = Path(title="The ID of the item to get", ge=0, le=1000),
    q: str,
    size: float = Query(gt=0, lt=10.5),
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

gt: greater than
ge: greater than or equal
lt: less than
le: less than or equal

Pass * as the first parameter of the function.

Python does nothing with that *, but it will know that all subsequent parameters should be treated as keyword parameters (key-value pairs), also known as kwargs, when called. Even if they do not have default values.

That is, if you do not add * as the first parameter, you cannot only fill in q: str in the function parameters because it does not have a default value.

# Multiple Model Parameters in Request Body#

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


class User(BaseModel):
    username: str
    full_name: Union[str, None] = None


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    results = {"item_id": item_id, "item": item, "user": user}
    return results

Subsequent parameters can use the following format.

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    }
}

## Single Value in Request Body#

If you want to add another key-value pair in the same request body,

For example,

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    },
    "key": "key"
}

Then you can introduce Body.

from fastapi import FastAPI, Query, Path, Body

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User, key: str = Body()):
    results = {"item_id": item_id, "item": item, "user": user, "key": key}
    return results

By adding the following code to achieve this.

key: str = Body()

Similarly, Body() can also add validation like Query and Path.

key: str = Body(gt=1)

Test interface.

## Embed Single Request Body Parameter#

If you want to achieve the following request format.

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

Instead of

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
}

You can use the embed parameter of Body().

For example,

item: Item = Body(embed=True)

from typing import Union

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


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


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item = Body(embed=True)):
    results = {"item_id": item_id, "item": item}
    return results

# Nested Models in Request Body#

## List Field with Subtypes#

First, import List.

from typing import List, Union

For example, if you need to pass multiple tags,

You can use the following code to represent List.

# Restrict the content type, can only pass str type
tags: List[str] = []

# No restriction on content type, str, int, etc. can all be passed
tags: List = []

The code is as follows.

from typing import List, Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: Union[float, None] = None
    tags: List[str] = []


@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    results = {"item_id": item_id, "item": item}
    return results

## Set Type#

Unlike List, elements in a Set cannot be duplicated, so it is more suitable for storing tags.

Because duplicate elements will be automatically merged.

You also need to import Set from typing.

from typing import Set

Using code.

tags: Set[str] = set()

The code is as follows.

class Message(BaseModel):
    role: str
    content: str
    tags: Set[str] = set()
  
@app.post("/chat")
async def chat(message: Message, model: str = Body()):
    return {"message": message, "model": model}

## Nested Models#

You can nest another model within a model.

For example,

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2,
    "tags": ["rock", "metal", "bar"],
    "image": {
        "url": "http://example.com/baz.jpg",
        "name": "The Foo live"
    }
}

First, define the sub-model.

class Image(BaseModel): 
    url: str 
    name: str

Then add it to the original model.

class Message(BaseModel):
    role: str
    content: str
    tags: Set[str] = set()
    image: Union[Image, None] = None

## Special Types and Validation#

In addition to ordinary single value types (such as str, int, float, etc.), you can also use more complex single value types that inherit from str.

External special types can be introduced via Pydantic, for details visit Pydantic Official Website.

For example, the URL in the image should be an address, and Pydantic provides the HttpUrl type for validation.

Similarly, first import.

from pydantic import BaseModel, HttpUrl

Then modify the image model class.

class Image(BaseModel):
    url: HttpUrl
    name: str

At this point, if the URL passed in does not start with http or https, an error will be raised.

## Properties with a Set of Sub-models#

Similarly, you can use Pydantic models as subtypes of list, set, etc.

For example,

class Message(BaseModel):
    role: str
    content: str
    tags: Set[str] = set()
    image: Union[List[Image], None] = None

# Response Models#

## User Registration Response#

You can use the response_model parameter in any path operation to declare the model used for the response:

For example, for user registration, after successful registration, return user data.

However, the password cannot be returned directly, so you can set a response model without a password.

Define the model for the registered user.

class User(BaseModel):
    username: str
    full_name: Union[str, None] = None
    email: EmailStr
    password: str

Define the response model returned after successful registration.

class UserOut(BaseModel):
    username: str
    full_name: Union[str, None] = None
    email: EmailStr

The implementation code is as follows.

@app.post("/user/", response_model=UserOut)
async def create_user(user: User) -> Any:
    return user

Here, Any means that the returned user can be of any type.

Test interface.

## Predefined Model Response#

First, define a response model.

class Item2(BaseModel):
    name: str
    description: Union[str, None] = None
    price: float
    tax: float = 10
    tags: List[str] = []

Then set the default return.

item2s = {
    "fire": {"name": "Fire", "price": 3.2},
    "peach": {"name": "Peach", "description": "Peach Peach Peach", "price": 2.2, "tags": ["aa", "bb", "cc"]},
    "firepeach": {"name": "firepeach", "description": "firepeach firepeach firepeach", "price": 2.2}
}

The interface implementation is as follows.

@app.get("/item2s/{item_id}", response_model=Item2)
async def read_item2(item_id: str):
    return item2s[item_id]

At this point, sending the item_id as a request parameter will return the corresponding data from item2s.

Test call.

If you do not want to return parameters with default values,

You can add a parameter response_model_exclude_unset=True.

Modify the code as follows.

@app.get("/item2s/{item_id}", response_model=Item2, response_model_exclude_unset=True)
async def read_item2(item_id: str):
    return item2s[item_id]

At this point, the interface call will not return data with default values or empty content.

Data that differs from the default value will still be returned.

# Additional Models#

When saving user data, you need to add a model for storing it in the database.

Because passwords cannot be saved in plain text, for example, they can be saved as hash values.

Add a UserDB model.

class UserDB(BaseModel):
    username: str
    hashed_password: str
    email: EmailStr
    full_name: Union[str, None] = None

Define a fake password hashing function.

def fake_password_hasher(raw_password: str):
    return "hashpassword" + raw_password

Define a function to create a UserDB model type of data and save it to the database.

def fake_save_user(user_in: User):
    fake_hash_password = fake_password_hasher(user_in.password)
    # You can use ** to unpack a dictionary into keyword arguments.
    # Then unpacked user_in as keyword arguments into user_db.
    # And assign the password hashed as hashed_password to user_db.
    user_db = UserDB(**user_in.dict(), hashed_password=fake_hash_password)

    print("User fake save successful!");
    return user_db

Add an interface for registering a user.

@app.post("/user2/", response_model=UserOut)
async def create_user2(user: User):
    user_save = fake_save_user(user)
    return user_save

Test interface.

## About `**user.dict()`#

### Pydantic Model#

user is a Pydantic model of the User class.

Pydantic models have a .dict() method that returns a dict containing the model data.

So if you create a User model object as follows:

user_in = User(username="john", password="secret", email="[email protected]")

If user.dict() is called,

user_dict = user.dict()

The data of user_dict will be:

{
    'username': 'john',
    'password': 'secret',
    'email': '[email protected]',
    'full_name': None,
}

That is, all properties are extracted.

### Unpacking dict#

If you pass a dict like user_dict to a function (or class) using **user_dict, Python will "unpack" it. It will pass the keys and values of user_dict as keyword arguments directly.

So if you use the following code:

UserDB(**user_dict)
# Or
UserDB(**user.dict())

The resulting UserDB will be:

UserDB(
    'username': 'john',
    'password': 'secret',
    'email': '[email protected]',
    'full_name': None,
)

Or more precisely, directly using user_dict to represent any content that may be included in the future:

UserDB(
    username = user_dict["username"],
    password = user_dict["password"],
    email = user_dict["email"],
    full_name = user_dict["full_name"],
)

Then add an additional keyword argument hashed_password=hashed_password, for example:

UserDB(**user.dict(), hashed_password=hashed_password)

The final result will be:

UserDB(
    username = user_dict["username"],
    password = user_dict["password"],
    email = user_dict["email"],
    full_name = user_dict["full_name"],
    hashed_password = hashed_password,
)

## About Code Duplication#

Code duplication increases the likelihood of bugs, security issues, and code drift (when you update code in one place but not in others).

The above models share a lot of data and have duplicate property names and types.

Therefore, based on the previously mentioned dict and unpacking, you can directly define a UserBase base class.

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None

And other models only need to assign new properties based on this.

For example,

class User(UserBase):
    password: str


class UserDB(UserBase):
    hashed_password: str


class UserOut(UserBase):
    # pass means no processing, keep consistent with UserBase
    pass

Because in the subsequent use of interfaces, most fields will have, only new fields need to be resolved.

## Union Model#

You can declare a response as two types using Union, which means that the response will be any of the two types.

This will be represented as anyOf in the documentation.

For example,

response_model=Union[PlaneItem, CarItem]

The code is as follows.

from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class BaseItem(BaseModel):
    description: str
    type: str

class CarItem(BaseItem):
    type = "car"

class PlaneItem(BaseItem):
    type = "plane"
    size: int

items = {
    "item1": {"description": "All my friends drive a low rider", "type": "car"},
    "item2": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "plane",
        "size": 5,
    },
}

@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
async def read_item(item_id: str):
    return items[item_id]

## Model List#

You can declare a response consisting of a list of objects in the same way.

response_model=List[Item]

The code is as follows.

from typing import List
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str


items = [
    {"name": "Foo", "description": "There comes my hero"},
    {"name": "Red", "description": "It's my aeroplane"},
]


@app.get("/items/", response_model=List[Item])
async def read_items():
    return items

# Response Status Codes#

You can specify the status code using status_code.

For example,

@app.post("/user2/", response_model=UserOut, status_code=201)
async def create_user2(user: User):
    user_saved = fake_save_user(user);
    return user_saved;

When successfully returned, it will return a status code of 201.

At the same time, FastAPI also provides convenient variables.

Import

from fastapi import status

@app.post("/user2/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def create_user2(user: User):
    user_saved = fake_save_user(user);
    return user_saved;

Common status codes include:

For more information about each status code and applicable scenarios, refer to HTTP Response Status Codes - HTTP | MDN.

# Form Data#

Unlike JSON, HTML forms (<form></form>) typically send data to the server using "special" encoding.

When receiving not JSON but form fields, you need to use Form.

To use forms, you need to install python-multipart.

Execute

pip install python-multipart

Import the Form provided by FastAPI.

from fastapi import FastAPI, Form
app = FastAPI()

Creating form (Form) parameters is done in the same way as Body and Query.

@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}

For example, the "password flow" mode of the OAuth2 specification requires sending username and password through form fields.

The specification requires that the fields must be named username and password, and sent through form fields, not JSON.

Using Form allows declaring the same metadata and validation as Body (and Query, Path, Cookie).

The media type encoding of form data is generally application/x-www-form-urlencoded.

But forms containing files are encoded as multipart/form-data.

# Request Files#

# Handling Errors#

In some cases, it is necessary to return error messages to the client.

The client referred to here includes front-end browsers, other applications, IoT devices, etc.

The scenarios where it is necessary to return error messages to the client mainly include:

The client does not have permission to perform the operation.
The client does not have permission to access the resource.
The item the client wants to access does not exist.
And so on...

In these cases, it is common to return 4XX (400 to 499) HTTP status codes.

4XX status codes are similar to 2XX (200 to 299) HTTP status codes that indicate a successful request.

However, 4XX status codes indicate errors that occurred on the client side.

## HTTPException#

To return HTTP error responses to the client, you can use HTTPException.

from fastapi import HTTPException

For example

@app.get("/item2s/{item_id}", response_model=Item2, response_model_exclude_unset=True)
async def read_item2(item_id: str):
    if item_id not in item2s:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such file")
    return item2s[item_id]

Test call.

## Installing Custom Exception Handlers#

To add custom handlers, use Starlette's exception tools.

Assuming the custom exception to be triggered is called UnicornException.

And you need FastAPI to implement global handling of this exception.

At this point, you can use @app.exception_handler() to add a custom exception controller:

You can also directly import FastAPI's encapsulated ones.

from fastapi import Request
from fastapi.responses import JSONResponse

The code is as follows.

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

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Ha! {exc.name} something went wrong..."},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name="Custom Error Name")
    return {"unicorn_name": name}

{"message": f"Ha! {exc.name} something went wrong..."} where f
can get the custom error name.

Test results.

# JSON Compatible Encoder and Updating Data#

In some cases, you may need to convert data types (such as Pydantic models) into JSON-compatible data types (such as dict, list, etc.).

For example, if you need to store it in a database.

For this requirement, FastAPI provides the jsonable_encoder() function.

## Using jsonable_encoder() to Update Data#

Assuming there is a database named fake_db, which only accepts JSON-compatible data.

For example, it does not accept objects like datetime because these objects are not compatible with JSON.

Therefore, datetime objects must be converted to str objects containing ISO formatting.

Similarly, this database will not accept Pydantic models (objects with properties), but only dict.

For this, you can use jsonable_encoder.

To update data, use HTTP PUT operation.

It accepts an object, such as a Pydantic model, and will return a JSON-compatible version:

Convert input data into data stored in JSON format (for example, when using NoSQL databases), you can use jsonable_encoder. For instance, converting datetime to str.

from datetime import datetime

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

# Fake database
fake_db = {
    "fee": {"title": "foo", "timestamp": datetime.now()}
}

# Define an ItemJ model
class ItemJ(BaseModel):
    title: str
    timestamp: datetime
    description: str | None = None

# put for updating
@app.put("/itemsJ/{id}")
def update_itemJ(id: str, item: ItemJ):
    # Convert time to str type in JSON format
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data
    return fake_db[id]

Test call.

# Security#

Assuming the backend API is in one domain.

The frontend is in another domain, or (in mobile applications) in a different path of the same domain.

And the frontend needs to use the backend's username and password to authenticate the user.

Certainly, FastAPI supports OAuth2 authentication.

## Overview#

Import Depends dependency and OAuth2PasswordBearer authentication.

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

At this point, in the API documentation at localhost:8000/docs, you can see

A "Authorize" button appears in the upper right corner of the page.

A small lock icon also appears in the upper right corner of the path operation.

Clicking the Authorize button pops up an authorization form to enter username and password and other optional fields:

If not, you need to install python-multipart.

Execute pip install python-multipart.

This is because OAuth2 uses form data to send `username` and `password`.

### Password Flow#

The Password flow is a way defined by OAuth2 to handle security and authentication (the flow).

The goal of OAuth2 is to allow the backend or API to be independent of server authentication of user identity.

But in this example, the FastAPI application will handle API and authentication.

Here is a simplified operational flow:

The user enters username and password in the frontend and presses Enter.
The frontend (running in the user's browser) sends the username and password to the URL specified in the API (declared with tokenUrl="token").
The API checks the username and password and responds with a token (Token) (this functionality is not yet implemented):
The token is just a string used to authenticate the user.
Generally, the token will expire after a period.
- After expiration, the user needs to log in again.
- This way, even if the token is stolen, the risk is lower. Because it is not a permanent key, in most cases it will not be valid for long.
The frontend temporarily stores the token somewhere.
The user clicks on the frontend to go to other parts of the frontend application.
The frontend needs to extract more data from the API:
- Authenticate for the specified endpoint (Endpoint).
- Therefore, when authenticating with the API, send a request header Authorization with the value Bearer + token.
- If the token is foobar, the Authorization request header would be: Bearer foobar.

### OAuth2PasswordBearer#

FastAPI provides security tools at different levels of abstraction.

This example uses the Password flow of OAuth2 and the Bearer token.

To do this, you need to use the OAuth2PasswordBearer class.

When creating an instance of OAuth2PasswordBearer, you need to pass the tokenUrl parameter. This parameter contains the URL for the client (the frontend running in the user's browser) to send username and password and obtain the token.

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

Here, tokenUrl="token" points to a relative URL token that has not yet been created. This relative URL is equivalent to ./token.

Because a relative URL is used, if the API is located at https://example.com/, it points to https://example.com/token. But if the API is located at https://example.com/api/v1/, it points to https://example.com/api/v1/token.

Using a relative URL is very important to ensure that the application can still run normally in advanced use cases such as behind a proxy.

### Usage#

The oauth2_scheme variable is an instance of OAuth2PasswordBearer and is callable.

The calling method is

oauth2_scheme(some, parameters)

Therefore, Depends (dependency) can call the oauth2_scheme variable.

Next, use Depends to pass oauth2_scheme as a dependency.

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

This dependency receives the parameter token of the path operation as a string (str).

## Get Current User#

### Create User Model#

class UserAuth(BaseModel):
    username: str
    # Use Union[str, None] first, will later change to EmailStr
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None

### Create Dependency#

Create a get_current_user dependency.

get_current_user will have the same oauth2_scheme as a dependency.

Just like what was done directly in the path operation, the new dependency get_current_user will receive a str type token from the sub-dependency oauth2_scheme:

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

### Get User Information#

The token obtained from oauth2_scheme will be used to get the user.

get_current_user will use the created (fake) utility function fake_decode_token, which takes a str type token and returns our Pydantic User model.

def fake_decode_token(token): 
    return User(username=token + "fakedecoded", email="[email protected]", full_name="John Doe")

### Inject Current User#

Now you can use get_current_user as Depends in path operations:

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

## Simple OAuth2 with Password and Bearer#

### Overview#

Now we will use FastAPI's security utilities to obtain username and password.

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

Moreover, the specification clearly states that the fields must be named this way. Therefore, user-name or email will not work.

However, the database model can use any other names.

But for the login path operation, these names need to be used to be compatible with the specification (to have the ability to use the integrated API documentation system).

The specification also states that username and password must be sent as form data (therefore, JSON cannot be used here).

### Get `username` and `password`#

#

Need to use OAuth2PasswordRequestForm.

First, import.

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_dict = fake_users_db.get(form_data.username)
    # If user_dict does not exist 
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    # Pass User data to UserDB model class
    user = UserDB(**user_dict)
    # Hash the password
    hashed_password = fake_hash_password(form_data.password)

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

    # Return username and token type
    return {"access_token": user.username, "token_type": "bearer"}

OAuth2PasswordRequestForm is a class dependency that declares the following request form:

username.
password.
An optional scope field, which is a large string consisting of space-separated strings.
An optional grant_type.

#### Password Hashing#

"Hashing" means converting something (in this case, the password) into a seemingly garbled byte sequence (just a string).

Every time you pass in exactly the same content (the exact same password), you will get exactly the same garbled result.

But you cannot convert the garbled result back to the password.

    # Hash the password
    hashed_password = fake_hash_password(form_data.password)

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

#### Return Token#

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

It should have a token_type. In this example, the "Bearer" token is used, so the token type should be "bearer".

And there should also be an access_token field, which is a string containing our access token.

For this simple example, we will extremely insecurely return the same username as the token.

    # Return username and token type
    return {"access_token": user.username, "token_type": "bearer"}

#### Update Dependency#

Now we will update the dependency.

We want to get the current_user only if this user is active.

So we create an additional dependency get_current_active_user, which in turn has get_current_user as a dependency.

If the user does not exist or is disabled, both dependencies will return an HTTP error.

Thus, in our endpoint, we can only get the user if they exist, authentication passes, and they are active:

# Get user information through token
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

# Get the current active user, active after logging in
# Only when authentication passes and the user is active can we get this user
async def get_current_active_user(current_user: UserAuth = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


# Create a fake database

fake_users_db = {
    "bryce": {
        "username": "bryce",
        "full_name": "bryce Yu",
        "email": "[email protected]",
        "hashed_password": "hash123123",
        "disabled": False,
    },

    "yu": {
        "username": "yu",
        "full_name": "yu",
        "email": "[email protected]",
        "hashed_password": "hash123456",
        "disabled": True,
    },
}

### Complete Code (Key Points!!!!)#

The complete code for user login to get the current user is as follows.

# Create a User model class
class UserAuth(BaseModel):
    username: str
    # Use Union[str, None] first, will later change to EmailStr
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None

# Create a User Database model class
class UserDB(UserAuth):
    hashed_password: str

# Simulate hashing the password
def fake_hash_password(raw_password: str):
    # Simulate hashing the password, should actually use bcrypt or other hashing algorithms
    return "hash" + raw_password

# Get user information through token
def fake_decode_token(token):
    # Get user information through token, currently insecurely using username as token
    user = get_user(fake_users_db, token)
    return user

# Get user information from the db database by username
def get_user(db, username: str):
    # Currently insecurely using username as token
    if username in db:
        user_dict = db[username]
        return UserDB(**user_dict)



# First, obtain the token through oauth2_scheme, then use the token to get the current logged-in user information
# oauth2_scheme is an authentication scheme that retrieves the value of the Authorization field from the request header and parses it into a token
async def get_current_user(token: str = Depends(oauth2_scheme)):
    # Get user information through token
    user = fake_decode_token(token)
    # If the user does not exist
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # If the user exists, return user information
    return user


# Get the current active user
async def get_current_active_user(current_user: UserAuth = Depends(get_current_user)):
    # Get the current logged-in user current_user through get_current_user
    # Then check if the user is disabled
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="User is disabled")
    return current_user


# Log in user through form
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):

    # Get the user from the simulated database by username
    user_dict = fake_users_db.get(form_data.username)

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

    # Pass User data to UserDB model class to get an instance of UserDB to compare passwords
    user = UserDB(**user_dict)

    # Hash the password to compare with the password in the database
    hashed_password = fake_hash_password(form_data.password)

    # If the password is incorrect
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    
    # Return token and token type, currently insecurely using username as token
    return {"access_token": user.username, "token_type": "bearer"}


# When calling this interface, you need to add Authorization: Bearer token in the headers, where token is the token returned during login
# When using openApi for testing, it will automatically add Authorization: Bearer token in the headers
# In actual requests, you need to manually add the Authorization value in the headers
@app.get("/users/me")
async def read_users_me(current_user: UserAuth = Depends(get_current_active_user)):
    # Get the current logged-in and not disabled user current_user through get_current_active_user
    return current_user

#### Process Summary#

First, call the /token interface, submit username and password through the form, and the interface will return a token.

Then, you need to add the value of the Token in the Authorization in the headers when calling the /user/me interface to get the current logged-in user.

Since the Bearer is used in this case, the value of Authorization is "Bearer " + the returned Token.

In the openApi testing of this case, the Token value will be automatically added to the Authorization.

In actual project usage, you need to manually add the Authorization value in the headers.