FastAPI

Contents

5.2. FastAPI#

5.2.1. Steps of Build Application#

5.2.1.1. Setup new environment#

%%cmd

# create new virtual environment
python -m venv fastapi_venv

# activate virtual environment
.\fastapi_venv\Scripts\activate

# active virtual environment macos
# source fastapi_venv/bin/activate

# (optional) deactive conda environment if you are using conda, then there is only one Python virtual environment is active.
conda deactivate

# install dependencies
pip install fastapi[standard] uvicorn

Microsoft Windows [Version 10.0.19045.4651]
(c) Microsoft Corporation. All rights reserved.

(base) c:\Users\datkt\Desktop\Working\00_My Notebooks\coding\learning\contents\3_programming_and_frameworks\python\python_frameworks\fastapi>python -m venv fastapi_venv

(base) c:\Users\datkt\Desktop\Working\00_My Notebooks\coding\learning\contents\3_programming_and_frameworks\python\python_frameworks\fastapi>

5.2.1.2. Magic Command to query API in notebook#

from IPython.core.magic import register_line_magic
import requests


# Register a new line magic command called query_api
@register_line_magic
def query_api(line):
    try:
        # Split the input line to get method, URL, and optional headers/data
        parts = line.split(" ")
        method = parts[0].upper()  # HTTP method (GET, POST, PUT, DELETE)
        url = parts[1]  # API URL

        # Optional parts (headers and payload can be passed as key=value format)
        headers = {}
        data = {}

        # Parse headers and data from the remaining parts (key=value format)
        for part in parts[2:]:
            if "=" in part:
                key, value = part.split("=")
                if key.startswith("header:"):
                    headers[key.replace("header:", "")] = value
                else:
                    data[key] = value

        # Make the appropriate request based on the method
        if method == "GET":
            response = requests.get(url, headers=headers, params=data)
        elif method == "POST":
            response = requests.post(url, headers=headers, json=data)
        elif method == "PUT":
            response = requests.put(url, headers=headers, json=data)
        elif method == "DELETE":
            response = requests.delete(url, headers=headers, json=data)
        else:
            return f"Unsupported method {method}"

        # Return the response content
        return (
            response.json() if response.status_code == 200 else response.text
        )

    except Exception as e:
        return str(e)


# Example Usage:
# %query_api GET https://jsonplaceholder.typicode.com/posts header:Accept=application/json
# %query_api POST https://jsonplaceholder.typicode.com/posts header:Content-Type=application/json title=foo body=bar userId=1
# %query_api PUT https://jsonplaceholder.typicode.com/posts/1 header:Content-Type=application/json title=foo body=bar userId=1
# %query_api DELETE https://jsonplaceholder.typicode.com/posts/1

magic command to write append to file

from IPython.core.magic import register_cell_magic


# Register a cell magic called appendfile
@register_cell_magic
def appendfile(line, cell):
    # The 'line' argument captures the line after the magic command (the file name)
    # The 'cell' argument captures the content between the magic command and the end of the cell

    filename = line.strip()  # Get the file name from the line input

    # Append the content of the cell to the file
    with open(filename, "a") as f:
        f.write(
            "\n" + cell + "\n"
        )  # Append the cell content and add a new line

    print(f"Appended to {filename}")


# Example usage:
# %%appendfile example.txt
# This is the content that will be appended.
# You can write as much content as needed in the cell.

5.2.1.3. Build the First App#

%%writefile main.py
from enum import Enum
from typing import Annotated, Literal

from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

# create a FastAPI "instance"
app = FastAPI()


# create a route theo cấu trúc @app.<operation>(path)
@app.get("/")
async def root():
    return {"message": "Hello World"}

Overwriting main.py

Run the app

%%cmd

# by FastAPI CLI in dev mode and reload if there are any changes of main.py
fastapi dev main.py

# run app in production mode
fastapi run main.py

Default address: http://127.0.0.1:8000

%query_api GET http://127.0.0.1:8000/

{'message': 'Hello World'}

5.2.1.4. Documentation#

API Documentation

The automatic interactive API documentation http://127.0.0.1:8000/docs (provided by Swagger UI) or http://127.0.0.1:8000/redoc (provided by ReDoc)

API Schema

This schema definition includes your API paths, the possible parameters they take, etc.

You can see it directly at: http://127.0.0.1:8000/openapi.json.

5.2.2. Cấu trúc của API#

5.2.2.1. Path (Endpoint)#

Path (hay còn gọi là Endpoint hoặc route) là đường dẫn mà client gửi yêu cầu tới API server. Đó là một phần của URL (Uniform Resource Locator) được sử dụng để chỉ định nguồn tài nguyên cụ thể.

Path chỉ định tài nguyên cụ thể

Ví dụ: Nếu API của bạn có URL là https://api.example.com/users/123, thì :

  • /users là Path. Đó là endpoint chỉ tới tài nguyên users (người dùng) trên server.

  • /users/123 là path truy cập vào tài nguyên người dùng có ID là 123

  • https://api.example.com là domain của server

5.2.2.4. Method (HTTP Methods)#

HTTP Methods (phương thức HTTP) xác định hành động mà client yêu cầu server thực hiện. Một số phương thức thông dụng trong API bao gồm:

  • GET: Lấy dữ liệu từ server (read data)

  • POST: Gửi dữ liệu đến server để tạo mới tài nguyên (create data)

  • PUT: Cập nhật toàn bộ tài nguyên đã tồn tại (update data)

  • PATCH: Cập nhật một phần tài nguyên (update apart of data)

  • DELETE: Xoá dữ liệu (delete data)

  • OPTIONS

  • HEAD

  • TRACE

HTTP Methods xác định hành động được thực hiện với tài nguyên.

Ví dụ:

  • GET yêu cầu:

  GET https://api.example.com/users/123

  • POST yêu cầu:

  POST https://api.example.com/users
  Body: {
    "name": "John",
    "email": "john@example.com"
  }

5.2.2.5. Body (Payload)#

Body chứa dữ liệu chính mà client muốn gửi tới server trong một yêu cầu HTTP, thường xuất hiện trong các yêu cầu có phương thức POST, PUT, PATCH, hoặc DELETE.

  • Dữ liệu trong body có thể là JSON, XML, hoặc các định dạng khác tuỳ thuộc vào ứng dụng.

Body chứa dữ liệu được gửi trong yêu cầu.

Ví dụ:

Trong một yêu cầu POST, body có thể chứa dữ liệu mà client muốn gửi đến server để tạo mới một đối tượng.

{
  "name": "John Doe",
  "email": "john@example.com"
}

Ở đây, body là dữ liệu JSON chứa thông tin người dùng mới.

5.2.2.6. Query (Query String)#

Query (hay Query Parameters) là các cặp giá trị key-value được thêm vào cuối URL để cung cấp thông tin cho server. Nó thường được sử dụng trong các yêu cầu GET để lọc, tìm kiếm, hoặc cung cấp các tham số bổ sung.

Query parameters được gắn vào URL sau dấu hỏi (?) và các cặp key-value được ngăn cách bởi dấu &.

Query là các tham số lọc hoặc tìm kiếm trong URL.

Ví dụ:

GET https://api.example.com/users?name=JohnDoe&age=30

  • Ở đây, name=JohnDoe và age=30 là các query parameters dùng để lọc danh sách người dùng theo tên và độ tuổi.

5.2.2.7. Status Codes#

Status Codes là mã số trong phản hồi của server dùng để cho biết trạng thái của yêu cầu. Một số mã phổ biến:

  • 200 OK: Yêu cầu thành công.

  • 201 Created: Tài nguyên được tạo thành công.

  • 400 Bad Request: Yêu cầu không hợp lệ.

  • 401 Unauthorized: Không có quyền truy cập, cần xác thực.

  • 404 Not Found: Không tìm thấy tài nguyên.

  • 500 Internal Server Error: Lỗi phía server.

Status Codes mô tả kết quả của yêu cầu.

5.2.2.8. Authentication và Authorization#

  • Authentication: Xác thực là quá trình xác định danh tính của người dùng. Nó đảm bảo rằng người gửi yêu cầu là người dùng hợp lệ.

    Ví dụ: Đăng nhập bằng tài khoản, mật khẩu.

  • Authorization: Phân quyền là quá trình xác định người dùng có quyền thực hiện các hành động cụ thể hay không. Sau khi xác thực, server kiểm tra quyền của người dùng với tài nguyên cụ thể.

5.2.2.9. Rate Limiting#

Rate Limiting là cơ chế hạn chế số lượng yêu cầu mà một người dùng hoặc một hệ thống có thể gửi tới API trong một khoảng thời gian cụ thể. Điều này giúp ngăn chặn việc quá tải server.

5.2.3. HTTP Method Operation#

API Operation

Each Path of API có thể đảm nhiệm 1 hoặc nhiều operation (HTTP method), có các loại operation như sau:

  • GET: Lấy dữ liệu từ server (read data)

  • POST: Gửi dữ liệu đến server để tạo mới tài nguyên (create data)

  • PUT: Cập nhật toàn bộ tài nguyên đã tồn tại (update data)

  • PATCH: Cập nhật một phần tài nguyên (update apart of data)

  • DELETE: Xoá dữ liệu (delete data)

  • OPTIONS

  • HEAD

  • TRACE

Sử dụng cấu trúc @app.<operation>(path) để decorate cho function tương ứng với path và operation đó:

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

  • Lưu ý ở đây là function có thể ở dạng normal là def hoặc bất đồng bộ là async def

  • Path và function phải là duy nhất và không được trùng lặp trong main.py

Model có thể return:

  • dict, list hoặc singular values as str, int, …

  • Pydantic model / JSON

  • …

5.2.4. Path parameters#

Path API định nghĩa là endpoint hoặc route : /path/<variable>

Ví dụ: trong đường dẫn https://example.com/items/foo thì path là /items/foo

Path can be designed with path parameters by using Python format string, khi đó giá trị của variable sẽ là 1 phần của đường dẫn

%%appendfile main.py


### Parameters Path: parameter in path
@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}


# testcase: http://127.0.0.1:8000/items/foo

Appended to main.py

Giá trị item_id sẽ được passed vào function như là 1 argument.

%query_api GET http://127.0.0.1:8000/items/foo

{'item_id': 'foo'}

5.2.4.1. Validate argument#

Có thể định nghĩa datatype bằng type hint hoặc các validators definition by pydantic. Khi có khi thực hiện function, các argument sẽ tự động được validate

%%appendfile main.py
### Parameters Path: parameter validation
from typing import Annotated
from pydantic import Field


@app.get("/items/{item_id}")
async def read_item_validation(item_id: Annotated[str, Field(max_length=5)]):
    return {"item_id": item_id}


# testcase: http://127.0.0.1:8000/items/foofoofoo ----> error

Appended to main.py

%query_api GET http://127.0.0.1:8000/items/foofoofoo

'{"detail":[{"type":"string_too_long","loc":["path","item_id"],"msg":"String should have at most 5 characters","input":"foofoofoo","ctx":{"max_length":5}}]}'

5.2.4.2. Pre-defined parameter values#

Định nghĩa trước các giá trị hợp lệ của parameters

%%appendfile main.py
### Parameters Path: Pre-defined parameters value
from enum import Enum


class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"


@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    # check model name in ModelName
    if model_name is ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}

    # or get value from arguments
    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}

    return {"model_name": model_name, "message": "Have some residuals"}


# testcase: http://127.0.0.1:8000/models/alexnet ---> Oke
# testcase: http://127.0.0.1:8000/models/lenaddd ---> Error

Appended to main.py

%query_api GET http://127.0.0.1:8000/models/alexnet

{'model_name': 'alexnet', 'message': 'Deep Learning FTW!'}

%query_api GET http://127.0.0.1:8000/models/lenaddd

'{"detail":[{"type":"enum","loc":["path","model_name"],"msg":"Input should be \'alexnet\', \'resnet\' or \'lenet\'","input":"alexnet_v2","ctx":{"expected":"\'alexnet\', \'resnet\' or \'lenet\'"}}]}'

5.2.4.3. Path containing paths#

Nếu trong trường hợp parameter là 1 path (cũng chứa ký tự “/”)

%%appendfile main.py


### Parameters Path: path in parameters
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}


# testcase: http://127.0.0.1:8000/files/documents/test.txt

Appended to main.py

%query_api GET http://127.0.0.1:8000/files/documents/test.txt

{'file_path': 'documents/test.txt'}

5.2.4.4. Multi path parameters#

Lấy nhiều biến từ path

%%appendfile main.py


### Parameters Path: multiple path parameters
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
    user_id: int,
    item_id: str,
):
    item = {"item_id": item_id, "owner_id": user_id}
    return item


# testcase: http://127.0.0.1:8000/users/1/items/foo

Appended to main.py

%query_api GET http://127.0.0.1:8000/users/1/items/foo

{'item_id': 'foo', 'owner_id': 1}

5.2.4.5. Sử dụng Path kết hợp với Annotated#

Chú ý: Path parameter là 1 required parameter, nên buộc phải khai báo, trong TH sử dụng default value hoặc None đều không có tác dụng

%%appendfile main.py
### Parameters Path: define Path parameter with Annotated + Path
from fastapi import Path


@app.get("/read_items_path_annotated/{item_id}")
async def read_items_path_annotated(
    item_id: Annotated[int, Path(title="The ID of the item to get")],
    q: Annotated[str | None, Query(alias="item-query")] = None,
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results


# testcase: http://127.0.0.1:8000/read_items_path_annotated/foo?item-query=dat

Appended to main.py

%query_api GET http://127.0.0.1:8000/read_items_path_annotated/1?item-query=dat

{'item_id': 1, 'q': 'dat'}

5.2.5. Query parameters#

Trong các parameter khác được khai báo không thuộc path parameter được định nghĩa thì chúng sẽ tự động hiểu là dạng Query Parameters. Cụ thể, Query parameters sẽ ở dạng key-value phía sau ? trong URL, được tác rời bởi &. Các tham số này sẽ là các tham số bổ sung cho function (ngoài path parameters nếu có).

Ví dụ: http://127.0.0.1:8000/items/?skip=0&limit=10 sẽ được query ra các biến:

  • skip = “0”

  • limit = “10”

Chúng ta hoàn toàn có thể định nghĩa data-type của biến bằng cách sử dụng type-hint hoặc pydantic validators Nếu không truyền biến trong URL, sẽ lấy giá trị default của biến

Ví dụ: ta truy cập đến path có cả path parameter và query parameters trong đó query variable có type là boolean

%%appendfile main.py


### Parameters Query: query parameters with boolean type
@app.get("/items_v2/{item_id}")
async def read_value(item_id: str, q: str | None = None, short: bool = False):
    item = {"item_id": item_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


# testcase: http://127.0.0.1:8000/items/foo?short=1    ----> short = True
# testcase: http://127.0.0.1:8000/items/foo?short=True ----> short = True
# testcase: http://127.0.0.1:8000/items/foo?short=true ----> short = True
# testcase: http://127.0.0.1:8000/items/foo?short=no  ---> short = False
# testcase: http://127.0.0.1:8000/items/foo?short=yes ---> short = True

Appended to main.py

%query_api GET http://127.0.0.1:8000/items/foo?short=1

{'item_id': 'foo'}

5.2.5.1. Add validation with Query#

Sử dụng fastapi.Query để add metadata cho Annotated (fastapi.Query is similar like pydantic.Field, but in Query parameter prefer to use Query)

Arguments of Query:

  • title

  • description

  • alias: sử dụng alias thay thế tên biến trong URL

  • deprecated: mark the parameter is on way of deprecated

%%appendfile main.py


### Parameters Query: use `Query` to add metadata for `Annotated`
@app.get("/read_items_query_validation/")
async def read_items_query_validation(
    q: Annotated[str | None, Query(min_length=3, max_length=5, default=None)],
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results


# testcase: http://127.0.0.1:8000/items/?q=foo ----> Oke
# testcase: http://127.0.0.1:8000/items/?q=fooooo ----> Error

Appended to main.py

%query_api GET http://127.0.0.1:8000/read_items_query_validation/?q=foo

{'items': [{'item_id': 'Foo'}, {'item_id': 'Bar'}], 'q': 'foo'}

%query_api GET http://127.0.0.1:8000/read_items_query_validation/?q=fooooo

'{"detail":[{"type":"string_too_long","loc":["query","q"],"msg":"String should have at most 5 characters","input":"fooooo","ctx":{"max_length":5}}]}'

5.2.5.2. Set Ellipsis (…) as required parameter#

Sử dụng ... as the default value for parameter but the python mark as like the required parameter

%%appendfile main.py


### Parameters Query: use `...` mark as the required value
@app.get("/required_items/")
async def read_required_items(q: Annotated[str, Query(min_length=3)] = ...):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results


# testcase: http://127.0.0.1:8000/required_items/ ----> Error

Appended to main.py

%query_api GET http://127.0.0.1:8000/required_items/

'Internal Server Error'

%query_api GET http://127.0.0.1:8000/required_items/?q=foo

{'items': [{'item_id': 'Foo'}, {'item_id': 'Bar'}], 'q': 'foo'}

5.2.5.3. Query parameters is list / multiple values#

Trong trường hợp biến truyền vào là 1 list, ta sử dụng nhiều lần khai báo biến ngăn cách bởi &

%%appendfile main.py


### Parameters Query: Query parameter is list / multiple values
@app.get("/items/")
async def read_items(q: Annotated[list[str] | None, Query()] = None):
    query_items = {"q": q}
    return query_items


# testcase: http://127.0.0.1:8000/items/?q=foo&q=bar

Appended to main.py

%query_api GET http://127.0.0.1:8000/items/?q=foo&q=bar

{'q': ['foo', 'bar']}

5.2.5.4. Alias parameters#

Thay vì sử dụng argument là tên biến trong function để đặt tên key trong URL thì sẽ sử dụng alias để thay thế

%%appendfile main.py


### Parameters Query: use `alias` as an alias for `Query`
@app.get("/items_alias/")
async def read_items_alias(
    q: Annotated[str | None, Query(alias="item-query")] = None,
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results


# testcase: http://127.0.0.1:8000/items_alias/?item-query=foo

Appended to main.py

%query_api GET http://127.0.0.1:8000/items_alias/?item-query=foo

{'items': [{'item_id': 'Foo'}, {'item_id': 'Bar'}], 'q': 'foo'}

5.2.5.5. Query with pydantic model#

Define the query parameter by pydantic model khi kết hợp với Query

Chú ý nếu parameter truyền vào là dạng pydantic model mà không được Annotated với Query thì parameter sẽ được hiểu là request body thay vì là query parameter

%%appendfile main.py


### Parameters Query: define the query parameter with pydantic model
class FilterParams(BaseModel):
    limit: int = Field(100, gt=0, le=100)
    offset: int = Field(0, ge=0)
    order_by: Literal["created_at", "updated_at"] = "created_at"
    tags: list[str] = []

    # set config for not allow unknown field
    model_config = {"extra": "forbid"}


@app.get("/read_items_pydantic/")
async def read_items_pydantic(filter_query: Annotated[FilterParams, Query()]):
    return filter_query


# testcase: http://127.0.0.1:8000/read_items_pydantic/?limit=10&tags=foo&tags=bar ---> Success
# testcase: http://127.0.0.1:8000/read_items_pydantic/?order_by=deleted_at ---> Error
# testcase: http://127.0.0.1:8000/read_items_pydantic/?var=10 ---> Error

Appended to main.py

%query_api GET http://127.0.0.1:8000/read_items_pydantic/?limit=10&tags=foo&tags=bar

{'limit': 10, 'offset': 0, 'order_by': 'created_at', 'tags': ['foo', 'bar']}

%query_api GET http://127.0.0.1:8000/read_items_pydantic/?order_by=deleted_at

'{"detail":[{"type":"literal_error","loc":["query","order_by"],"msg":"Input should be \'created_at\' or \'updated_at\'","input":"deleted_at","ctx":{"expected":"\'created_at\' or \'updated_at\'"}}]}'

%query_api GET http://127.0.0.1:8000/read_items_pydantic/?var=10

'{"detail":[{"type":"extra_forbidden","loc":["query","var"],"msg":"Extra inputs are not permitted","input":"10"}]}'

5.2.6. 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.

Your API almost always has to send a response body. But clients don’t necessarily need to send request bodies all the time, sometimes they only request a path, maybe with some query parameters, but don’t send a body.

The request body is often sent by client in POST, PUT, DELETE PATCH method

5.2.6.1. Request body by pydantic model (+ query + path)#

Parameter theo thứ tự nhận diện như sau:

  1. Nếu parameter được định nghĩa trong path thì sẽ sử dụng nó như path parameter

  2. Nếu parameter là singular type (like int, float, str, bool, etc) thì parameter được xác định như query parameter

  3. Nếu parameter được định nghĩa theo dạng pydantic model thì sẽ hiểu là request body

%%appendfile main.py


### Parameters Body | POST: use pydantic model (as request body) + path + query parameters
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


@app.post("/items_pydantic/{item_id}/")
async def create_item_pydantic_model(
    item_id: int, item: Item, q: str | None = None
):
    item_dict = item.model_dump()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    result = {"item_id": item_id, **item_dict}
    if q:
        result.update({"q": q})
    return result

Appended to main.py

5.2.6.2. Multiple body parameters#

Sử dụng nhiều body parameters vào request

%%appendfile main.py


### Parameters Body | PUT: multiple body parameters
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


class User(BaseModel):
    username: str
    full_name: str | None = None


@app.post("/items_multiple_body/{item_id}")
async def update_items_multiple_body(item_id: int, item: Item, user: User):
    results = {"item_id": item_id, "item": item, "user": user}
    return results


# testcase: http://127.0.0.1:8000/items_multiple_body/1/?q=foo header:Content-Type=application/json item

Appended to main.py

body = {
    "item": {"name": "a", "description": "asdasd", "price": 123, "tax": 23},
    "user": {"username": "name", "full_name": "fname"},
}

5.2.6.3. Singular value in body#

Thay vì sử dụng 1 dict hoặc 1 object cho body, ta chỉ cần 1 value, khi đó ta sử dụng Body (có thể kết hợp với Annotated)

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


class User(BaseModel):
    username: str
    full_name: str | None = None


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

Khi đó body sẽ là:

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

5.2.6.4. Embed a single body to multiple body#

Nếu trong hàm cần sử dụng multiple body, nhưng ta chỉ muốn khi truyền vào thì body là 1 JSON thay vì truyền nhiều body, thì có thể sử dụng Body(embed=True), khi đó body parameter là value có key là argument.

%%appendfile main.py


### Parameters Body | POST: emmbed body
class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


class User(BaseModel):
    username: str
    full_name: str | None = None


@app.put("/items_embed_body/{item_id}")
async def update_items_embed_body(
    item_id: int,
    item: Annotated[Item, Body(embed=True)],
    user: Annotated[User, Body(embed=True)],
    importance: Annotated[int, Body()],
):
    results = {
        "item_id": item_id,
        "item": item,
        "user": user,
        "importance": importance,
    }
    return results

5.2.6.5. Add metadata with Field#

Sử dụng pydantic.Field để add metadata cho variables

5.2.6.6. Nested model in body#

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: set[str] = set()
    image: Image | None = None  # Nested pydantic model


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

The expected body is

{
    "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"
    }
}

5.2.7. Config Router#

5.2.7.1. Response model type#

Ta có thể chỉ định Output của model bằng TypeHint tuy nhiên trong một số TH sẽ không phù hợp, do đó phương pháp thay thế là sử dụng response_model.

Thông qua ví dụ:

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from typing import Any

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str | None = None


class UserOut(BaseModel):
    username: str
    email: EmailStr


# Nếu sử dụng output là UserIn thì sẽ gửi password cho client
@app.post("/user/")
async def create_user(user: UserIn) -> UserIn:
    return user


# Nếu sử dụng output là UserOut thì sẽ bị lỗi trong quá trình validate do UserIn và UserOut là 2 model khác nhau
@app.post("/user/")
async def create_user_diff_out(user: UserIn) -> UserOut:
    return user


# Thay vào đó, nên sử dụng `response_model`
@app.post("user/", response_model=UserOut)
async def create_user_resmodel(user: UserIn) -> Any:
    return user

5.2.7.2. Response Encoding - filtering#

5.2.7.2.1. Loại bỏ các field có giá trị default value#

  1. response_model_exclude_unset=True

Chú ý: khác với việc là có giá trị truyền vào, và giá trị đó bằng với giá trị default value

5.2.7.2.2. Lựa chọn trường tuỳ chỉnh#

  • response_model_include: set of str with the name of the attributes to include

  • response_model_exclude: set of str with the name of the attributes to exclude

If you forget to use a set and use a list or tuple instead, FastAPI will still convert it to a set and it will work correctly

  • response_model_exclude_unset=True: including the ones that were not set and have their default values.

    This is different from response_model_exclude_defaults in that if the fields are set, they will be included in the response, even if the value is the same as the default.

  • response_model_exclude_defaults=True: including the ones that have not the same value as the default

  • response_model_exclude_none=True: exclude fields set to None

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float = 10.5


items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {
        "name": "Bar",
        "description": "The Bar fighters",
        "price": 62,
        "tax": 20.2,
    },
    "baz": {
        "name": "Baz",
        "description": "There goes my baz",
        "price": 50.2,
        "tax": 10.5,
    },
}


@app.get(
    "/items/{item_id}/name",
    response_model=Item,
    response_model_include={"name", "description"},
)
async def read_item_name(item_id: str):
    return items[item_id]


@app.get(
    "/items/{item_id}/public",
    response_model=Item,
    response_model_exclude={"tax"},
)
async def read_item_public_data(item_id: str):
    return items[item_id]

5.2.7.3. Response status code#

Định nghĩa cụ thể / Điều chỉnh Status Code nhận được khi response trả về

Mô tả status code : detail

1. Information (100 và cao hơn)

  • 100-199: Các mã này cho biết rằng yêu cầu đã được nhận và đang được xử lý. Thường không được sử dụng trực tiếp trong phản hồi.

2. Successful (200 và cao hơn)

  • 200 OK: Yêu cầu đã thành công.

  • 201 Created: Một tài nguyên đã được tạo thành công (thường được sử dụng sau các yêu cầu POST).

  • 204 No Content: Yêu cầu đã thành công, nhưng không có nội dung để trả về.

3. Redirection (300 và cao hơn)

  • 300 Multiple Choices: Cho biết có nhiều tùy chọn cho tài nguyên.

  • 301 Moved Permanently: Tài nguyên đã được chuyển vĩnh viễn đến một URL mới.

  • 302 Found: Tài nguyên đã được chuyển tạm thời.

  • 304 Not Modified: Tài nguyên không có thay đổi kể từ yêu cầu trước; không có nội dung nào nên được trả về.

4. Client Errors (400 và cao hơn)

  • 400 Bad Request: Yêu cầu không thể được hiểu bởi máy chủ do cú pháp sai.

    Ví dụ: Khi người dùng gửi dữ liệu đầu vào không hợp lệ cho mô hình: Yêu cầu gửi một JSON không đúng định dạng

  • 401 Unauthorized: Cần xác thực và xác thực đã thất bại hoặc chưa được cung cấp.

    Ví dụ: Khi người dùng không cung cấp token xác thực: Người dùng cố gắng truy cập API mà không có token.

  • 403 Forbidden: Máy chủ hiểu yêu cầu nhưng từ chối cấp quyền.

    Ví dụ: Khi người dùng cố gắng truy cập vào tài nguyên không có quyền: Người dùng không có quyền truy cập vào mô hình nhạy cảm.

  • 404 Not Found: Tài nguyên yêu cầu không thể được tìm thấy.

    Ví dụ: Khi người dùng cố gắng truy cập một mô hình không tồn tại: Yêu cầu đến URL /models/12345 nhưng mô hình với ID 12345 không tồn tại.

5. Server Errors (500 và cao hơn)

  • 500 Internal Server Error: Lỗi chung đã xảy ra trên máy chủ.

    Ví dụ: Khi có lỗi xảy ra trong quá trình xử lý yêu cầu: Mô hình gặp sự cố trong quá trình dự đoán do lỗi trong mã.

  • 502 Bad Gateway: Máy chủ nhận được phản hồi không hợp lệ từ máy chủ phía trên.

  • 503 Service Unavailable: Máy chủ hiện không thể xử lý yêu cầu do quá tải tạm thời hoặc bảo trì.

    Ví dụ: Khi dịch vụ AI/ML đang bảo trì: Người dùng gửi yêu cầu trong khi hệ thống đang được bảo trì.

from fastapi import FastAPI, status

app = FastAPI()


# short
@app.post("/items_shortcode/", status_code=201)
async def create_item_shortcode(name: str):
    return {"name": name}


# full
@app.post("/items_fullcode/", status_code=status.HTTP_201_CREATED)
async def create_item_fullcode(name: str):
    return {"name": name}

5.2.7.4. Endpoint Information#

Bổ sung thông tin của endpoint

  • summary: Mô tả endpoint

  • description: Mô tả chi tiết action của endpoint

  • tags: đánh label dùng để phân loại, có thể thuộc nhiều loại: tags = ['items','user']

  • deprecated: đánh dấu sắp xoá bỏ

  • response_description: Mô tả response

from enum import Enum

from fastapi import FastAPI

app = FastAPI()


# đánh tags bằng Enum
class Tags(Enum):
    items = "items"
    users = "users"


@app.get("/items/", tags=["items"])
async def get_items():
    return ["Portal gun", "Plumbus"]


@app.get("/users/", tags=[Tags.users])
async def read_users():
    return ["Rick", "Morty"]


@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",
    description="Create an item with all the information, name, description, price, tax and a set of unique tags",
    tags=[Tags.items],
    deprecated=True,
    response_description="The created item",
)
async def create_item(item: Item):
    return item

5.2.7.5. Function paramter description#

Mô tả bằng docstring

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


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


@app.post(
    "/create_item_docstring/", response_model=Item, summary="Create an item"
)
async def create_item_docstring(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item

5.2.7.6. Import Router từ submodule#

Trong các dự án lớn thì thường mỗi 1 nhóm routers sẽ là 1 file python riêng (thay vì để tất cả trong file app.py). Khi đó trong mỗi file router, ta sẽ định nghĩa các endpoint của nhóm routers đó, sau đó add vào file app tại file chính.

Chi tiết hãy xem ví dụ : Example Big Application

.
├── app                  # "app" is a Python package
│   ├── __init__.py      # this file makes "app" a "Python package"
│   ├── main.py          # "main" module, e.g. import app.main
│   ├── dependencies.py  # "dependencies" module, e.g. import app.dependencies
│   └── routers          # "routers" is a "Python subpackage"
│   │   ├── __init__.py  # makes "routers" a "Python subpackage"
│   │   ├── items.py     # "items" submodule, e.g. import app.routers.items
│   │   └── users.py     # "users" submodule, e.g. import app.routers.users
│   └── internal         # "internal" is a "Python subpackage"
│       ├── __init__.py  # makes "internal" a "Python subpackage"
│       └── admin.py     # "admin" submodule, e.g. import app.internal.admin

5.2.7.6.1. Config trực tiếp trong từng router tại submodule#

Với mỗi router/endpoint:

  • Tự định nghĩa path mẹ (ví dụ \users\)

  • Tự đánh tags

  • …

%%writefile routers/users.py

from fastapi import APIRouter

router = APIRouter()


@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]


@router.get("/users/me", tags=["users"])
async def read_user_me():
    return {"username": "fakecurrentuser"}


@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}

Khi import vào file app chính

%%writefile main.py

from fastapi import Depends, FastAPI
from routers import items, users

app = FastAPI()
app.include_router(users.router)

5.2.7.6.2. Config chung cho all router tại submodule#

Với all router/endpoint. ta sẽ đặt những config chung thay vì phải set lặp lại các config chung cho từng router:

  • Set path mẹ chung cho all router (ví dụ \items\)

  • Set tags chung

  • …

Những config riêng của từng router thì sẽ set như cách 1 phía trên, và nó sẽ bổ sung cho config chung

%%writefile routers/items.py

from fastapi import APIRouter, Depends, HTTPException

from dependencies import get_token_header

# config chung cho tất cả các routers
router = APIRouter(
    prefix="/items", # tất cả các router khi chạy ở main đều có path đứng đầu là "items"
    tags=["items"], # tất cả các config đều có tag là items
    dependencies=[Depends(get_token_header)], # Extra responses
    responses={404: {"description": "Not found"}}, # they all need that X-Token dependency we created.
)


fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}


@router.get("/")
async def read_items():
    return fake_items_db


@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"name": fake_items_db[item_id]["name"], "item_id": item_id}


@router.put(
    "/{item_id}",
    tags=["custom"], # add more tags
    responses={403: {"description": "Operation forbidden"}}, # add more responses
)
async def update_item(item_id: str):
    if item_id != "plumbus":
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus"
        )
    return {"item_id": item_id, "name": "The great Plumbus"}

Khi import vào file app chính

%%writefile main.py

from fastapi import Depends, FastAPI
from routers import users

app = FastAPI()
app.include_router(users.router)

5.2.7.6.3. Config khi import vào app chính#

Với cách này, mỗi router/endpoint chỉ cần đặt những config riêng của từng endpoint mà bỏ qua những điểm chung, những config chung này thay vì config khi tạo APIRouter thì sẽ config khi add router vào app chính tại main.py

%%writefile routers/admin.py

from fastapi import APIRouter

router = APIRouter()


@router.post("/") # không cần setup path là "/admin/"
async def update_admin():
    return {"message": "Admin getting schwifty"}

Khi import vào file app chính

Ta sẽ config khi add router vào app

%%writefile main.py

from fastapi import Depends, FastAPI
from internal import admin

app = FastAPI()
# config routers when add to app in main 
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)

5.2.8. Cookie#

Cookie là tệp nhỏ được server gửi và lưu trữ trên trình duyệt người dùng. Khi người dùng truy cập lại, cookie được gửi lại server để lưu trữ thông tin như phiên làm việc, trạng thái đăng nhập, hoặc tuỳ chọn cá nhân.

Coockie sẽ là 1 tệp nhỏ được lưu truyển qua lại giữa client và server nhằm xác định client

Ứng dụng trong FastAPI cho AI/ML:

  • Lưu trạng thái phiên làm việc: Giúp theo dõi phiên làm việc của người dùng khi sử dụng API, ví dụ để lưu lịch sử dự đoán hoặc thông tin xác thực.

  • Quản lý token xác thực: Cookie có thể lưu token xác thực (JWT) khi người dùng đăng nhập vào ứng dụng ML, giúp API xác minh người dùng mà không cần nhập lại thông tin.

from fastapi import FastAPI, Response, Cookie, HTTPException

app = FastAPI()


# /login: Lưu token xác thực vào cookie.
@app.post("/login")
def login(response: Response):
    token = "secure_token"

    # Set giá trị `auth_token` trong cookie và biến này sẽ được lưu trong phiên làm việc
    response.set_cookie(key="auth_token", value=token)
    return {"message": "Logged in"}


# /predict: Kiểm tra cookie để xác nhận người dùng hợp lệ trước khi trả về kết quả dự đoán từ model.
@app.get("/predict")
def predict(
    # lấy giá trị `auth_token` trong cookie tự động
    # Nếu trong Cookie không có biến auth_token thì trả giá trị None
    auth_token: str = Cookie(None),
):
    if auth_token != "secure_token":
        raise HTTPException(status_code=403, detail="Unauthorized")
    return {"prediction": "Model output"}

Setup Cookie by Pydantic Model

Extract giá trị cookie nhận được bên trong request vào Cookies Model được được defined trước

from typing import Annotated

from fastapi import Cookie, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Cookies(BaseModel):
    # config: chỉ nhận các giá trị được defined phía dưới
    # If a client tries to send some extra cookies, they will receive an error response.
    model_config = {"extra": "forbid"}

    session_id: str
    fatebook_tracker: str | None = None
    googall_tracker: str | None = None


@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):
    return cookies

5.2.9. Header#

Header trong API là phần của yêu cầu (request) hoặc phản hồi (response) HTTP, chứa các thông tin bổ sung về yêu cầu hoặc phản hồi đó. Header giúp giao tiếp giữa client và server hiệu quả hơn, cho phép truyền tải thông tin như loại nội dung, độ dài nội dung, và các thông tin xác thực.

Vai trò của Header:

  • Xác thực và phân quyền: Header thường được sử dụng để gửi các token xác thực như JWT (JSON Web Tokens) để xác định danh tính người dùng và quyền truy cập vào các tài nguyên.

  • Chỉ định loại nội dung: Bạn có thể chỉ định loại nội dung mà client muốn nhận (ví dụ: Accept: application/json) hoặc loại nội dung mà server gửi (ví dụ: Content-Type: application/json).

  • Không muốn lưu trữ thông tin lâu dài: Nếu thông tin chỉ cần trong một yêu cầu duy nhất, sử dụng header là hợp lý.

Header thường được sử dụng cho các thông tin tạm thời và để quản lý xác thực một cách linh hoạt.

from typing import Annotated

from fastapi import FastAPI, Header
from pydantic import BaseModel

app = FastAPI()


class CommonHeaders(BaseModel):
    model_config = {"extra": "forbid"}

    host: str
    save_data: bool
    if_modified_since: str | None = None
    traceparent: str | None = None
    x_tag: list[str] = []


@app.get("/read_items_header/")
async def read_items_header(headers: Annotated[CommonHeaders, Header()]):
    return headers

5.2.10. Dependency injection#

Dependency injection (sử dụng Depends) là một tính năng mạnh mẽ cho phép bạn tách rời và tái sử dụng logic trong các hàm xử lý (route handlers), hàm khởi tạo, hoặc bất kỳ thành phần nào khác.

5.2.10.1. Tái sử dụng code#

Khi bạn có một logic xử lý phức tạp như kiểm tra phân quyền, kết nối cơ sở dữ liệu, khai báo parameter thường dùng hoặc tính toán một giá trị nào đó, bạn có thể định nghĩa những logic này trong các hàm độc lập. Sau đó, bạn chỉ cần “inject” (chèn) những hàm này vào các hàm xử lý khác nhờ vào Depends. Điều này giúp code dễ bảo trì và dễ tái sử dụng.

from fastapi import Depends, FastAPI

app = FastAPI()


def get_token():
    return "some-token"


async def common_parameters(
    q: str | None = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items_depend(
    commons: Annotated[dict, Depends(common_parameters)],
    token: str = Depends(get_token),
):
    return {"token": token}, commons

Trong ví dụ này, hàm get_token() được sử dụng làm dependency cho route read_items. Mỗi khi route này được gọi, FastAPI sẽ tự động gọi hàm get_token() và lấy giá trị trả về làm tham số token cho route.

5.2.10.2. Quản lý vòng đời và trạng thái#

Dependency injection cũng có thể được sử dụng để quản lý vòng đời của các đối tượng, ví dụ như các đối tượng kết nối cơ sở dữ liệu hoặc phiên (session). Điều này đảm bảo các đối tượng đó được tạo ra, sử dụng, và hủy đúng lúc.

from fastapi import Depends


def get_db():
    db = "DB Connection"
    try:
        yield db
    finally:
        print("Closing DB Connection")


@app.get("/users/")
async def get_users(db=Depends(get_db)):
    return {"db": db}

Ở đây, FastAPI đảm bảo rằng kết nối tới cơ sở dữ liệu (giả sử là một chuỗi “DB Connection”) sẽ được tạo và hủy một cách an toàn sau khi kết thúc yêu cầu.

5.2.10.3. Xử lý phân quyền (authentication/authorization)#

Dependency injection thường được sử dụng để xử lý các vấn đề về phân quyền. Bạn có thể tạo các hàm kiểm tra phân quyền như là các dependency, và sử dụng chúng ở bất kỳ đâu trong ứng dụng.

from fastapi import Depends, HTTPException  # noqa: F811

VALID_TOKENS = [
    "token1",
    "token2",
    ]

def verify_token(token: str):
    if token in VALID_TOKENS:
        raise HTTPException(status_code=400, detail="Invalid token")


@app.get("/secure-data/")
async def read_secure_data(token: str = Depends(verify_token)):
    return {"secure_data": "This is secure"}

Hàm verify_token kiểm tra token được truyền vào. Nếu không hợp lệ, sẽ trả về lỗi HTTP. Dependency này đảm bảo mọi yêu cầu tới route đều được kiểm tra phân quyền.

5.2.10.4. Dependencies with yield#

Use dependencies to do some extra steps after finish by use with yield instead of return and write the extra steps code after this.

Make sure, yeild is used one single time per dependency

A database dependency with yield

Use this to create a database session and close it after finishing

async def get_db():
    
    # Only the code Before to and yield statement is executed before creating a response
    db = DBSession()
    try:
        yield db
        
    # The code following the yield statement is executed after the response has been delivered
    finally:
        db.close()

5.2.11. Background Tasks#

Background Tasks is tasks would be run after returning a response

Sử dụng cho các tác vụ cần sử lý sau khi nhận được request từ Client, nhưng Client không cần phải chờ cho đến khi tác vụ đó hoàn thành để nhận được response

Ví dụ: Khi bạn cần thực hiện các tác vụ tốn thời gian như gửi email, xử lý file, cập nhật cơ sở dữ liệu mà không muốn chờ kết quả.

  • Gửi email xác nhận đăng ký hoặc quên mật khẩu.

  • Ghi log hoặc lưu trữ dữ liệu không quan trọng sau khi trả về phản hồi.

  • Xử lý file upload hoặc tải về mà không ảnh hưởng đến thời gian phản hồi của API.

  • You receive a file that must go through a slow process, you can return a response of “Accepted” (HTTP 202) and process the file in the background.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

# define background task
def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    # `write_notification` is set as a background task, send email and message as arguments
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

Background tasks work with dependency injection

  • write_log is defined as a background tasks

  • When calling the send_notification function, before the message is written to the log and sent to email, it will check to see if the query exists or not? If so, it will be writen in the log

from typing import Annotated

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, 
    background_tasks: BackgroundTasks, 
    q: Annotated[str, Depends(get_query)] # check q is existed and write log first
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

5.2.12. Middleware#

Middleware là các thành phần trung gian nằm giữa yêu cầu (request) của người dùng và phản hồi (response) từ app.

Chỉnh sửa Middleware cho phép bạn can thiệp và xử lý các bước trước khi một yêu cầu được gửi đến endpoint và sau khi phản hồi được tạo ra từ endpoint.

Middleware Use-case Example:

  • Xử lý các request từ HTTP: áp dụng thêm steps cho tất cả các yêu cầu HTTP

  • CORS (Cross-Origin Resource Sharing): Cho phép hoặc chặn các yêu cầu từ các nguồn khác nhau.

  • TrustedHostMiddleware: Chỉ cho phép các domain hoặc host nhất định truy cập vào ứng dụng.

  • GZipMiddleware: Nén phản hồi của ứng dụng sử dụng GZip để tăng hiệu suất.

5.2.12.1. Middleware: Xử lý các request từ HTTP#

@app.middleware("http"): Đây là decorator dùng để xác định một middleware cho FastAPI. Middleware này được áp dụng cho tất cả các yêu cầu HTTP.

Sau khi define thì tất cả các hàm xử lý request HTTP sẽ phải chạy qua hàm middleware này

  • Xử lý các yêu cầu: Như kiểm tra, xác thực, hoặc thay đổi các giá trị của yêu cầu trước khi nó được xử lý bởi các router hoặc endpoint.

  • Xử lý các phản hồi: Thêm header, thay đổi nội dung phản hồi, hoặc thực hiện các bước logging sau khi phản hồi đã được tạo ra.

  • Logging: Ghi lại thông tin về các yêu cầu và phản hồi, thời gian xử lý.

  • Xác thực (Authentication): Kiểm tra token hoặc cookie để đảm bảo người dùng đã được xác thực.

Cách hoạt động

Middleware nhận vào 2 argument là request và call_next, trong đó:

  • request: Đối tượng đại diện cho yêu cầu HTTP:

  • call_next: Hàm được sử dụng để gọi tiếp đến các thành phần khác trong chuỗi xử lý (ví dụ: endpoint hoặc middleware khác).

Thứ tự thực thi của middleware

Khi bạn khai báo nhiều middleware, chúng sẽ được thực thi theo một thứ tự nhất định, và thứ tự này phụ thuộc vào vị trí chúng được khai báo trong code.

  1. Khi nhận yêu cầu (request): Middleware sẽ được thực thi theo thứ tự từ trên xuống dưới trong mã nguồn. Middleware đầu tiên được khai báo sẽ xử lý yêu cầu trước, sau đó chuyển tiếp đến middleware thứ hai, và cứ tiếp tục như vậy cho đến middleware cuối cùng.

  2. Khi trả phản hồi (response): Quá trình trả phản hồi diễn ra theo thứ tự ngược lại. Middleware cuối cùng sẽ xử lý phản hồi trước, sau đó chuyển tiếp lại qua các middleware trước nó, và cuối cùng là đến middleware đầu tiên.

import time

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    # Tạo bộ đếm thời gian - logging
    start_time = time.perf_counter()
    
    # Verify token / User
    token = request.headers.get("Authorization")
    if token != "Bearer my_secure_token":  # Kiểm tra token giả định
        raise HTTPException(status_code=401, detail="Unauthorized")
    
    # Thêm một thuộc tính vào yêu cầu
    request.state.custom_value = "This is custom data"
    
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Kết quả: Với mỗi yêu cầu HTTP gửi đến ứng dụng, middleware này sẽ:

  1. Ghi lại thời gian bắt đầu khi nhận yêu cầu.

  2. Gọi endpoint hoặc middleware tiếp theo.

  3. Tính toán thời gian xử lý toàn bộ yêu cầu (bao gồm thời gian xử lý trong các middleware và endpoint).

  4. Thêm một header có tên là “X-Process-Time” vào phản hồi, trong đó giá trị của nó là thời gian xử lý của yêu cầu.

Thứ tự run middleware:

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

app = FastAPI()

# Middleware A
@app.middleware("http")
async def middleware_a(request: Request, call_next):
    print("Middleware A - Before")
    response = await call_next(request)
    print("Middleware A - After")
    return response

# Middleware B
@app.middleware("http")
async def middleware_b(request: Request, call_next):
    print("Middleware B - Before")
    response = await call_next(request)
    print("Middleware B - After")
    return response

# Middleware C
@app.middleware("http")
async def middleware_c(request: Request, call_next):
    print("Middleware C - Before")
    response = await call_next(request)
    print("Middleware C - After")
    return response

# create a route theo cấu trúc @app.<operation>(path)
@app.get("/")
async def root():
    return {"message": "Hello World"}

client = TestClient(app)

response = client.get("/")

Middleware C - Before
Middleware B - Before
Middleware A - Before
Middleware A - After
Middleware B - After
Middleware C - After

5.2.12.2. CORS (Cross-Origin Resource Sharing)#

CORSMiddleware: Cho phép hoặc chặn các yêu cầu từ các nguồn khác nhau.

CORS (Chia sẻ tài nguyên giữa các nguồn gốc khác nhau) là cơ chế cho phép một trang web ở nguồn này (domain, giao thức, cổng khác nhau) có thể yêu cầu tài nguyên từ một nguồn khác. Khi một ứng dụng cần giao tiếp với các tài nguyên từ một domain khác, ví dụ như một API từ server khác, trình duyệt sẽ thực hiện kiểm tra CORS để xác định có cho phép việc truy cập đó không.

CORSMiddleware trong FastAPI cho phép bạn cấu hình CORS, như cho phép hoặc chặn các yêu cầu từ những nguồn khác nhau.

  • allow_origins - A list of origins that should be permitted to make cross-origin requests. E.g. ['https://example.org', 'https://www.example.org']. You can use ['*'] to allow any origin.

  • allow_origin_regex - A regex string to match against origins that should be permitted to make cross-origin requests. e.g. 'https://.*\.example\.org'.

  • allow_methods - A list of HTTP methods that should be allowed for cross-origin requests. Defaults to ['GET']. You can use ['*'] to allow all standard methods.

  • allow_headers - A list of HTTP request headers that should be supported for cross-origin requests. Defaults to []. You can use ['*'] to allow all headers. The Accept, Accept-Language, Content-Language and Content-Type headers are always allowed for simple CORS requests.

  • allow_credentials - Indicate that cookies should be supported for cross-origin requests. Defaults to False. Also, allow_origins cannot be set to ['*'] for credentials to be allowed, origins must be specified.

  • expose_headers - Indicate any response headers that should be made accessible to the browser. Defaults to [].

  • max_age - Sets a maximum time in seconds for browsers to cache CORS responses. Defaults to 600.

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Cấu hình CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],  # Chỉ cho phép các yêu cầu từ domain example.com
    allow_credentials=True,
    allow_methods=["GET", "POST"],  # Chỉ cho phép các phương thức GET, POST
    allow_headers=["*"],  # Cho phép tất cả các header
)

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

  • Ở đây, chỉ các yêu cầu từ https://example.com mới được phép.

  • Các phương thức HTTP được phép là GET và POST.

  • Middleware này cũng hỗ trợ credentials (ví dụ cookie) trong yêu cầu CORS.

Kết quả là nếu có yêu cầu từ nguồn không nằm trong danh sách cho phép (allow_origins), yêu cầu đó sẽ bị chặn.

5.2.12.3. TrustedHostMiddleware: Giới hạn truy cập domain/host#

Chỉ cho phép các domain hoặc host nhất định truy cập vào ứng dụng

TrustedHostMiddleware giúp tăng cường bảo mật cho ứng dụng bằng cách kiểm tra Host header của các yêu cầu. Điều này có thể giúp ngăn chặn một số cuộc tấn công như DNS Rebinding (kỹ thuật tấn công lợi dụng sự không tin cậy của DNS để lấy thông tin từ hệ thống cục bộ).

from fastapi import FastAPI
from starlette.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()

# Cấu hình TrustedHostMiddleware
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.org"],  # Chỉ cho phép các domain cụ thể
)

@app.get("/")
async def read_root():
    return {"message": "This host is trusted!"}

  • Middleware này chỉ cho phép các yêu cầu từ các domain example.com và các subdomain của example.org (dùng dấu * để chỉ các subdomain).

  • Nếu yêu cầu đến từ một domain không nằm trong danh sách này, yêu cầu sẽ bị từ chối.

Middleware này giúp bạn bảo vệ ứng dụng khỏi các yêu cầu từ những host không đáng tin cậy, nâng cao tính bảo mật.

5.2.12.4. GZipMiddleware: Nén response#

Nén phản hồi của ứng dụng sử dụng GZip để tăng hiệu suất

GZipMiddleware giúp nén phản hồi của ứng dụng trước khi trả về cho client. Nén dữ liệu giúp giảm kích thước phản hồi và tăng tốc độ truyền dữ liệu qua mạng, đặc biệt hữu ích khi trả về các phản hồi lớn như file JSON hoặc HTML.

from fastapi import FastAPI
from starlette.middleware.gzip import GZipMiddleware

app = FastAPI()

# Thêm GZip middleware để nén phản hồi
app.add_middleware(GZipMiddleware, minimum_size=1000)

@app.get("/")
async def read_root():
    return {"message": "This response is not compressed because it's small."}

@app.get("/large")
async def read_large():
    return {"message": "X" * 10000}  # Nén phản hồi lớn với chuỗi 10000 ký tự

  • minimum_size=1000: Phản hồi chỉ được nén nếu kích thước của nó lớn hơn 1000 byte. Phản hồi nhỏ sẽ không bị nén để tránh tốn tài nguyên xử lý.

  • Phản hồi từ endpoint /large sẽ được nén vì nó có độ dài lớn hơn 1000 byte.

Giảm kích thước phản hồi HTTP để cải thiện hiệu suất, đặc biệt hữu ích cho các ứng dụng gửi dữ liệu lớn như API hoặc trang web nhiều nội dung.

5.2.13. Security#

Cơ chế bảo mật của FastAPI

Do based trên OpenAPI nên FastAPI thừa kế security flow của OpenAPI

  • apiKey: chỉ là key mà thôi, có thể đến từ query param, header hoặc cookie.

  • http: hệ thống xác thực của HTTP, bao gồm:

    • bearer: header param với giá trị là một token (thừa kế từ OAuth2)

    • HTTP Basic authentication

    • HTTP Digest authentication

  • oauth2

  • openIdConnect

Các chuẩn bảo mật

1. OAuth 2

Là 1 chuẩn giao thức ủy quyền ra đời vào tháng 10 năm 2012, được sử dụng ở hầu hết mọi ứng dụng (web, mobile), cho phép người dùng cung cấp thông tin cá nhân bởi ứng dụng của bên thứ 3, cũng được dùng để cung cấp cơ chế cho việc xác thực người dùng.

2. OAuth 1

3. OpenID

4. OpenAPI

Example: Build bằng OAuth2

Trong hàm main.py, thêm cơ chế xác thực dựa trên login username/password của OAuth2.

Mỗi một endpoint thêm Depend nhằm xác thực user

from typing import Annotated

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

app = FastAPI()

# Tạo 1 instance của OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# truyền param vào hàm read_items biến token có type là string phụ thuộc instance oauth2_scheme
@app.get("/items/")
async def read_items(token: Annotated[str, Depends(oauth2_scheme)]):
    return {"token": token}
# --> Param này sẽ ngó qua request ở Authorization header, kiểm tra xem có token hay không và trả về, nếu không có thì sẽ báo lỗi 401 UNAUTHORIZED.

Khi chạy app, kiểm tra trong doc ta có 1 url api /items/, góc phải trên cùng có 1 box Authorize là nơi để điền username/password.

Áp dụng trong thực tế khi có frontend:

  • Người dùng sẽ đăng nhập username/password trên frontend

  • Frontend (browser chẳng hạn) gửi username và password tới url api (tokenUrl="token")

  • API kiểm tra username và password, trả về kết quả là 1 token (code trên chưa có đoạn này, đoạn dưới sẽ đề cập)

    • Token là 1 chuỗi các ký tự dùng để xác thực người dùng.

    • Token tồn tại có thời hạn, thời hạn này do bên phía backend setup

  • Frontend sau đó lưu token ở một nơi tạm thời nào đó (cookie, ram, …)

Validate khi người dùng điền username/password

from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel

# Giả sử dang sách username/password có sẵn
users = {
    "admin": {"username": "admin", "password": "123456"},
    "client1": {"username": "user1", "password": "123456"},
}

app = FastAPI()

# Tạo 1 instance của OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    username = form_data.username
    for user_id, user in users.items():
        if username == user["username"]:
            if not form_data.password == user["password"]:
                raise HTTPException(status_code=400, detail="Incorrect username or password")
            break
    else: # run all users
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    return {"access_token": user["username"], "token_type": "bearer"}

5.2.14. Set Example in parameter#

5.2.14.1. Example in pydantic model#

Edit model_config each pydantic model

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                }
            ]
        }
    }


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

5.2.14.2. Example in Field#

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class Item(BaseModel):
    name: str = Field(examples=["Foo"])
    description: str | None = Field(
        default=None, examples=["A very nice Item"]
    )
    price: float = Field(examples=[35.4])
    tax: float | None = Field(default=None, examples=[3.2])


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

5.2.14.3. Example in OpenAPI: Path(), Query(), …#

When using any of:

  • Path()

  • Query()

  • Header()

  • Cookie()

  • Body()

  • Form()

  • File()

Declare a group of examples with additional information that will be added to their JSON Schemas inside of OpenAPI.

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None


@app.put("/update_item_openapi_schema/{item_id}")
async def update_item_openapi_schema(
    item_id: int,
    item: Annotated[
        Item,
        Body(
            examples=[
                {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                },
                {
                    "name": "Bar",
                    "price": "35.4",
                },
            ],
        ),
    ],
):
    results = {"item_id": item_id, "item": item}
    return results

5.2.15. Testing#

Tạo file app.py

from typing import Annotated

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

fake_secret_token = "coneofsilence"

fake_db = {
    "foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
    "bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}

app = FastAPI()


class Item(BaseModel):
    id: str
    title: str
    description: str | None = None


@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item.id in fake_db:
        raise HTTPException(status_code=409, detail="Item already exists")
    fake_db[item.id] = item
    return item

tạo file test: test_app_client.py

  • import app từ file app.py

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_item():
    response = client.get("/items/foo", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 200
    assert response.json() == {
        "id": "foo",
        "title": "Foo",
        "description": "There goes my hero",
    }


def test_read_item_bad_token():
    response = client.get("/items/foo", headers={"X-Token": "hailhydra"})
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_read_nonexistent_item():
    response = client.get("/items/baz", headers={"X-Token": "coneofsilence"})
    assert response.status_code == 404
    assert response.json() == {"detail": "Item not found"}


def test_create_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},
    )
    assert response.status_code == 200
    assert response.json() == {
        "id": "foobar",
        "title": "Foo Bar",
        "description": "The Foo Barters",
    }


def test_create_item_bad_token():
    response = client.post(
        "/items/",
        headers={"X-Token": "hailhydra"},
        json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"},
    )
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid X-Token header"}


def test_create_existing_item():
    response = client.post(
        "/items/",
        headers={"X-Token": "coneofsilence"},
        json={
            "id": "foo",
            "title": "The Foo ID Stealers",
            "description": "There goes my stealer",
        },
    )
    assert response.status_code == 409
    assert response.json() == {"detail": "Item already exists"}

Thực hiện test bằng thư viện pytest, lệnh này sẽ tự động chạy tất cả các hàm bắt đầu bằng test_ trong file test.py, bao gồm các test case đã viết.

# pip install pytest
pytest test.py

5.2.16. Debugging - Development#

FastAPI trong chế độ development mode (dev mode) giúp phát hiện các lỗi trong quá trình phát triển và tự động cập nhật ứng dụng khi bạn thay đổi mã nguồn

  • Chạy bằng FastAPI CLI

fastapi dev main.py
# or
fastapi dev main.py --port 8000 --host 0.0.0.0

  • Chạy bằng unicorn

uvicorn main:app --reload
# or
uvicorn main:app --reload --port 8000 --host 0.0.0.0

  • Chạy bằng python

Yêu cầu trong main.py có khởi chạy unicorn

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

Run code in CMD: python main.py

5.2.17. Deployment#

  • Chạy bằng FastAPI CLI

fastapi run main.py --port 8000 --host 0.0.0.0

  • Chạy bằng unicorn

uvicorn main:app --port 8000 --host 0.0.0.0

  • Chạy bằng python

Yêu cầu trong main.py có khởi chạy unicorn

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="127.0.0.1", port=8000)

Run code in CMD: python main.py

Contents