[PYTHON] How to use FastAPI ① Tutorial --User Guide

What is written

--Memo when using FastAPI (It may be wrong because it is a personal memo ...) --Implemented by following the official website documentation (Tutorial --User Guide)

reference

FastAPI

environment

Build a Docker environment and check the operation

Docker FastAPI

Start-up

bash


# main.If py is in the root directory
$ uvicorn main:app --reload --host 0.0.0.0 --port 8000

# main.If py is not in the root directory
$ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

First Step

Interactive API Docs

Swagger

Alternative API docs

ReDoc

Check OpenAPI before automatic document generation

openapi.json

Simple Fast API functionality

python.main.py


from fastapi import FastAPI # ①

    app = FastAPI() # ②

    @app.get("/") # ③
        def read_root(): # ④
            return {"message": "Hello World"} # ⑤

① Import the required packages ② Create a Fast API instance ③ Operate with Path Operation Decorator and specify the path

operation(Method) Description
POST add to
GET Get
PUT update
DELETE Delete

(4) Specify the function to be called when receiving a request with Path Operation Function

⑤ Return the content (JSON)

Path Parameters

Basic

■

main.py


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

■Request

http://localhost:8000/items/1

■Response

{
  "item_id": "1"
}

Type declaration / validation

If a parameter that is not the specified type is passed, an error will be returned automatically.

■Source

main.py


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

■Request

http://localhost:8000/items/1

■Response

{
  "item_id": 1
}

■Request

http://localhost:8000/items/test

■Response

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

Path Operation Decorator order

If there is a URL that you want to take precedence over other path parameters, pay attention to the order. If you write / users / me later, the path parameter will take precedence.

■Source

main.py


@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}

■Request

http://localhost:8000/users/me

■Response

{
user_id: "the current user"
}

■Request

http://localhost:8000/users/user

■Response

{
user_id: "user"
}

Predefined path parameters

It is possible to predefine valid path parameters by using Enum.

■Source

main.py


from fastapi import FastAPI
from enum import Enum

app = FastAPI()

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

@app.get("/model/{model_name}")
async def get_model(model_name: ModelName):
    if model_name == ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}

    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}

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

■Request

http://localhost:8000/model/alexnet

■Response

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

http://localhost:8000/model/test

{
  detail: [
    {
      loc: [
        "path",
        "model_name"
      ],
      msg: "value is not a valid enumeration member; permitted: 'alexnet', 'resnet', 'lenet'",
      type: "type_error.enum",
      ctx: {
        enum_values: [
          "alexnet",
          "resnet",
          "lenet"
        ]
      }
    }
  ]
}

###Path parameters including path

In the path parameterpathIt is possible to receive the pass by describing.

■Source

main.py


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

■Request

http://localhost:8000/files/home/johndoe/myfile.txt

■Response

main.py


{
  "file_path": "/files/home/johndoe/myfile.txt"
}

##Query Parameters

If a parameter that is not a path parameter is passed, it will be automatically determined as a query parameter.

###initial value

■Source

main.py


from fastapi import FastAPI

app = FastAPI()

fake_item_db = [
    {"name": "name1"},
    {"name": "name2"},
    {"name": "name3"},
    {"name": "name4"},
    {"name": "name5"},
    {"name": "name6"},
    {"name": "name7"},
    {"name": "name8"},
    {"name": "name9"},
    {"name": "name10"}
]

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

■Request

http://localhost:8000/items

■Response

[
  {
    name: "name1"
  },
  {
    name: "name2"
  },
  {
    name: "name3"
  }
]

■Request

http://localhost:8000/items?skip=3

■Response

[
  {
    name: "name4"
  },
  {
    name: "name5"
  },
  {
    name: "name6"
  }
]

■Request

http://localhost:8000/items?limit=5

■Response

[
  {
    name: "name1"
  },
  {
    name: "name2"
  },
  {
    name: "name3"
  },
  {
    name: "name4"
  },
  {
    name: "name5"
  }
]

■Request

http://localhost:8000/items?skip=3&limit=4

■Response

[
  {
    name: "name4"
  },
  {
    name: "name5"
  },
  {
    name: "name6"
  },
  {
    name: "name7"
  }
]

###option

■Source

main.py


from typing import Optional
from fastapi import FastAPI

app = FastAPI()

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

■Request

http://localhost:8000/items/test

■Response

{
  "item_id": "test"
}

■Request

http://localhost:8000/items/test?q=aaa

■Response

{
  "item_id": "test",
  "q": "aaa"
}

###Type declaration

For boolean type1、True、on、yesAlso specifiedtrueIs judged.

###Source

main.py


@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Optional[str] = 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

■Request

http://localhost:8000/items/test

■Response

{
  "item_id": "test",
  "description": "This is an amazing item that has a long description"
}

■Request

http://localhost:8000/items/test?short=true

■Response

{
  "item_id": "test"
}

###Multiple paths and query parameters

■Source

main.py


@app.get("/users/{user_id}/items/{item_id}")
async def read_item(user_id: int, item_id: str, q: Optional[str] = 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

■Request

http://localhost:8000/users/1/items/test

■Response

{
  "item_id": "test",
  "owner_id": 1,
  "description": "This is an amazing item that has a long description"
}

###Required query parameters

■Source

main.py


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

■Request

http://localhost:8000/items/1?needy=test

■Response

{
  "item_id": "1",
  "needy": "test"
}

■Request

http://localhost:8000/items/1

■Response

{
  detail: [
    {
      loc: [
        "query",
        "needy"
      ],
      msg: "field required",
      type: "value_error.missing"
    }
  ]
}

##Request Body

###Request body generation by data model

■Source

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel

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

app = FastAPI()

@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

http://localhost:8000/items

{
  "name": "test",
  "price": 1000
}

■Response

{
  "name": "test",
  "description": null,
  "price": 1000,
  "tax": null
}

■Request

http://localhost:8000/items

{
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100
}

■Response

{
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100,
  "price_with_tax": 1100
}

###Request body + path parameter

■Source

main.py


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

■Request

http://localhost:8000/items/1

{
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100
}

■Response

{
  "item_id": 1,
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100
}

###Request body + path parameter + query parameter

■Source

main.py


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

■Request

http://localhost:8000/items/1?q=test

{
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100
}

■Response

{
  "item_id": 1,
  "name": "test",
  "description": "test test test",
  "price": 1000,
  "tax": 100,
  "q": "test"
}

##Query Parameters and String Validations

It is possible to set validation and metadata for query parameters.

###Verification / optional parameters

■Source

main.py


from fastapi import FastAPI, Query
from typing import Optional

app = FastAPI()

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

■Request

http://localhost:8000/items/?q=0123456789

■Response

{
  "items": [
    {
      "item_id": "Foo"
    },
    {
      "item_id": "Bar"
    }
  ],
  "q": "0123456789"
}

■Request

http://localhost:8000/items/?q=012345678901234567890

■Response

{
    detail: [
        {
            loc: [
                "query",
                "q"
            ],
            msg: "ensure this value has at most 20 characters",
            type: "value_error.any_str.max_length",
            ctx: {
                limit_value: 20
            }
        }
    ]
}

###Verification / required parameters

■Source

main.py


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

■Request

http://localhost:8000/items/?q=12345

■Response

{
  "items": [
    {
      "item_id": "Foo"
    },
    {
      "item_id": "Bar"
    }
  ],
  "q": "12345"
}

###Verification / regular expression

■Source

main.py


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

■Request

http://localhost:8000/items/?q=test12

■Response

{
  "items": [
    {
      "item_id": "Foo"
    },
    {
      "item_id": "Bar"
    }
  ],
  "q": "test12"
}

■Request

http://localhost:8000/items/?q=tes12

■Response

{
    detail: [
        {
            loc: [
                "query",
                "q"
            ],
            msg: "string does not match regex "^test"",
            type: "value_error.str.regex",
            ctx: {
                pattern: "^test"
            }
        }
    ]
}

###List query parameters(No default)

■Source

main.py


from fastapi import FastAPI, Query
from typing import List, Optional

app = FastAPI()

@app.get("/items/")
async def read_items(q: Optional[List[str]] = Query(None)):
    query_items = {"q": q}
    return query_items

■Request

http://localhost:8000/items/?q=123&q=456&q=test

■Response

{
  "q": [
    "123",
    "456",
    "test"
  ]
}

###List query parameters(With default)

■Source

main.py


from fastapi import FastAPI, Query
from typing import List, Optional

app = FastAPI()

@app.get("/items/")
async def read_items(q: Optional[List[str]] = Query(["foo", "bar"])):
    query_items = {"q": q}
    return query_items

■Request

http://localhost:8000/items

■Response

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

###Metadata settings

The set metadata is automatically reflected in the OpenAPI document. deprecatedIt can be clearly stated that it is deprecated by using. aliasBy using, it is possible to separate the variable name in the function from when receiving it as a path parameter.

■Source

main.py


@app.get("/items/")
async def read_items(q: Optional[str] = Query(
        None,
        min_length=5,
        max_length=20,
        title="metadata_title",
        description="metadata_description",
        deprecated=True,
        alias="alias_q"
    )):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results
img1.jpg img2.jpg

##Path Parameters and Numeric Validations

###Metadata settings

It is possible to set metadata in the path parameter as well.

■Source

main.py


from typing import Optional
from fastapi import FastAPI, Path, Query

app = FastAPI()

@app.get("/items/{item_id}")
async def read_items(
    item_id: int = Path(
      ...,
      title="metadata_title",
      description="metadata_description"
    ),
    q: Optional[str] = Query(
        None,
        min_length=5,
        max_length=20,
        title="metadata_title",
        description="metadata_description",
        deprecated=True,
        alias="alias_q"
    )):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

■Request

http://localhost:8000/items/1

■Response

{
  "item_id": 1
}

###Numerical validation

■Source

Comparison operator Description
ge that's all
gt large
le Less than
lt small

main.py


@app.get("/items/{item_id}")
async def read_items(*, item_id: int = Path(..., ge=1), q: str):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

■Request

http://localhost:8000/items/1?q=test

■Response

{
  "item_id": 1,
  "q": "test"
}

##Body - Multiple Parameters

###Combination of path parameters, query parameters and request body

■Source

main.py


from typing import Optional
from fastapi import FastAPI, Path
from pydantic import BaseModel

app = FastAPI()

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

@app.put("/items/{item_id}")
async def update_item(
    *,
    item_id: int = Path(..., title="The ID of the item to get", ge=0, le=1000),
    q: Optional[str] = None,
    item: Optional[Item] = None,
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    if item:
        results.update({"item": item})
    return results

■Request

http://localhost:8000/items/1

{
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  }
}

###Multiple request bodies

■Source

main.py


from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

class User(BaseModel):
    username: str
    full_name: Optional[str] = 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

■Request

http://localhost:8000/items/1

{
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  }
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  }
}

###Explicit specification of request body

If you specify a parameter in the function that is not a path parameter, FastAPI considers it a query parameter. Explicitly if you want the request body instead of the query parameter= Body(...)Is described.

■Source

main.py


from typing import Optional
from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()

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

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

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

■Request

http://localhost:8000/items/1

{
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  },
  "importance": 0
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  },
  "importance": 0
}

###Multiple request bodies and query parameters

In addition to the explicit request body, query parameters can also be specified.

■Source

main.py


from typing import Optional
from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()

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

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

@app.put("/items/{item_id}")
async def update_item(
    *,
    item_id: int,
    item: Item,
    user: User,
    importance: int = Body(..., gt=0),
    q: Optional[str] = None
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    if q:
        results.update({"q": q})
    return results

■Request

http://localhost:8000/items/1

{
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  },
  "importance": 1
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0
  },
  "user": {
    "username": "string",
    "full_name": "string"
  },
  "importance": 1
}

##Body - Fields

###Set meta tags for each field in the request body

By setting a meta tag in the schema, it will be automatically reflected in OpenAPI.

■Source

main.py


from typing import Optional
from fastapi import Body, FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = Field(
        None,
        title="The description of the item",
        max_length=300
    )
    price: float = Field(
        ...,
        gt=0,
        description="The price must be greater than zero"
    )
    tax: Optional[float] = 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
img3.jpg

##Body - Nested Models

###Set list in request body field

■Source

main.py


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

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = 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

■Request

http://localhost:8000/items/1

{
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0,
  "tags": [
    "foo",
    "bar"
  ]
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0,
    "tags": [
      "foo",
      "bar"
    ]
  }
}

###Nested model

■Source

main.py


from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

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

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

■Request

http://localhost:8000/items/1

{
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0,
  "tags": [
    "string"
  ],
  "image": {
    "url": "https://google.com",
    "name": "Google"
  }
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0,
    "tags": [
      "string"
    ],
    "image": {
      "url": "https://google.com",
      "name": "Google"
    }
  }
}

###List of submodels

■Source

main.py


from typing import List, Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()

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

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []
    images: Optional[List[Image]] = None

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

■Request

http://localhost:8000/items/1

{
  "name": "string",
  "description": "string",
  "price": 0,
  "tax": 0,
  "tags": [
    "string"
  ],
  "images": [
    {
      "url": "https://google.com",
      "name": "Google"
    },
    {
      "url": "https://yahoo.com",
      "name": "Yahoo"
    }
  ]
}

■Response

{
  "item_id": 1,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0,
    "tags": [
      "string"
    ],
    "images": [
      {
        "url": "https://google.com",
        "name": "Google"
      },
      {
        "url": "https://yahoo.com",
        "name": "Yahoo"
      }
    ]
  }
}

###Deeply nested model

■Source

main.py


from typing import List, Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()

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

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []
    images: Optional[List[Image]] = None

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

@app.post("/offers/")
async def create_offer(offer: Offer):
    return offer

■Request

http://localhost:8000/offers/

{
  "name": "string",
  "description": "string",
  "price": 0,
  "items": [
    {
      "name": "string",
      "description": "string",
      "price": 0,
      "tax": 0,
      "tags": [
        "string"
      ],
      "images": [
        {
          "url": "https://google.com",
          "name": "Google"
        },
        {
          "url": "https://yahoo.com",
          "name": "Yahoo"
        }
      ]
    },
    {
      "name": "string",
      "description": "string",
      "price": 0,
      "tax": 0,
      "tags": [
        "string"
      ],
      "images": [
        {
          "url": "https://google.com",
          "name": "Google"
        },
        {
          "url": "https://yahoo.com",
          "name": "Yahoo"
        }
      ]
    }
  ]
}

■Response

{
  "name": "string",
  "description": "string",
  "price": 0,
  "items": [
    {
      "name": "string",
      "description": "string",
      "price": 0,
      "tax": 0,
      "tags": [
        "string"
      ],
      "images": [
        {
          "url": "https://google.com",
          "name": "Google"
        },
        {
          "url": "https://yahoo.com",
          "name": "Yahoo"
        }
      ]
    },
    {
      "name": "string",
      "description": "string",
      "price": 0,
      "tax": 0,
      "tags": [
        "string"
      ],
      "images": [
        {
          "url": "https://google.com",
          "name": "Google"
        },
        {
          "url": "https://yahoo.com",
          "name": "Yahoo"
        }
      ]
    }
  ]
}

###List of models

■Source

main.py


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

app = FastAPI()

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

@app.post("/images/multiple/")
async def create_multiple_images(images: List[Image]):
    return images

■Request

http://localhost:8000/images/multiple/

[
  {
    "url": "https://google.com",
    "name": "Google"
  },
  {
    "url": "https://yahoo.com",
    "name": "Yahoo"
  }
]

■Response

[
  {
    "url": "https://google.com",
    "name": "Google"
  },
  {
    "url": "https://yahoo.com",
    "name": "Yahoo"
  }
]

##Schema Extra - Example

###Setting schema example in Config

■Source

main.py


from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

    class Config:
        schema_extra = {
            "example": {
                "name": "Foo",
                "description": "A very nice Item",
                "price": 35.4,
                "tax": 3.2,
            }
        }

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

###Setting a schema example for each field of the model

■Source

main.py


from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

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

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

###Setting schema example in function

■Source

main.py


from typing import Optional
from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()

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

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item = Body(
        ...,
        example={
            "name": "Foo",
            "description": "A very nice Item",
            "price": 35.4,
            "tax": 3.2,
        },
    ),
):
    results = {"item_id": item_id, "item": item}
    return results
img4.jpg

##Extra Data Types

■Source

main.py


from datetime import datetime, time, timedelta
from typing import Optional
from uuid import UUID
from fastapi import Body, FastAPI

app = FastAPI()

@app.put("/items/{item_id}")
async def read_items(
    item_id: UUID,
    start_datetime: Optional[datetime] = Body(None),
    end_datetime: Optional[datetime] = Body(None),
    repeat_at: Optional[time] = Body(None),
    process_after: Optional[timedelta] = Body(None),
):
    start_process = start_datetime + process_after
    duration = end_datetime - start_process
    return {
        "item_id": item_id,
        "start_datetime": start_datetime,
        "end_datetime": end_datetime,
        "repeat_at": repeat_at,
        "process_after": process_after,
        "start_process": start_process,
        "duration": duration,
    }

■Request

http://localhost:8000/items/3fa85f64-5717-4562-b3fc-2c963f66afa6

{
  "start_datetime": "2020-07-25T09:00:08.265Z",
  "end_datetime": "2020-07-25T09:00:08.265Z",
  "repeat_at": "18:20:30",
  "process_after": 0
}

■Response

{
  "item_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
  "start_datetime": "2020-07-25T09:00:08.265000+00:00",
  "end_datetime": "2020-07-25T09:00:08.265000+00:00",
  "repeat_at": "18:20:30",
  "process_after": 0,
  "start_process": "2020-07-25T09:00:08.265000+00:00",
  "duration": 0
}

##Header Parameters

###Use of header

■Source

main.py


from typing import Optional
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):
    return {"User-Agent": user_agent}

■Request

http://localhost:8000/items/

■Response

{
  "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.89 Safari/537.36"
}

###Duplicate header

■Source

main.py


from typing import List, Optional
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(x_token: Optional[List[str]] = Header(None)):
    return {"X-Token values": x_token}

■Request

bash


curl -X GET "http://localhost:8000/items/" -H  "accept: application/json" -H  "x-token: foo,bar"

■Response

{
  "X-Token values": [
    "foo,bar"
  ]
}

##Response Model

###Set an output model different from the input

■Source

main.py


from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

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

@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user

■Request

http://localhost:8000/user/

{
  "username": "user",
  "password": "pass123",
  "email": "[email protected]",
  "full_name": "aiueo"
}

■Response

{
  "username": "user",
  "email": "[email protected]",
  "full_name": "aiueo"
}

###Exclude the return of default values

Normally, all Fields specified in the model are returned. response_model_exclude_nusetOnly the Field for which the value is set is returned by using.

■Source

main.py


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

app = FastAPI()

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

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):
    return items[item_id]

■Request

http://localhost:8000/items/foo

■Response

{
  "name": "Foo",
  "price": 50.2
}

###Specify the field to return

It is possible to specify the Field to be returned from the model.

■Source

main.py


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

app = FastAPI()

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

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

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

■Request

http://localhost:8000/items/bar

■Response

{
  "name": "Bar",
  "description": "The bartenders"
}

##Extra Models

###Utilization of multiple models

■Source

main.py


from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

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

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

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

def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db

@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    return user_saved

■Request

http://localhost:8000/user/

{
  "username": "string",
  "password": "string",
  "email": "string",
  "full_name": "string"
}

■Response

{
  "username": "string",
  "email": "string",
  "full_name": "string"
}

###Eliminate model duplication

Eliminate duplication by using model inheritance.

■Source

main.py


from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

class UserIn(UserBase):
    password: str

class UserOut(UserBase):
    pass

class UserInDB(UserBase):
    hashed_password: str

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

def fake_save_user(user_in: UserIn):
    hashed_password = fake_password_hasher(user_in.password)
    user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)
    print("User saved! ..not really")
    return user_in_db

@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):
    user_saved = fake_save_user(user_in)
    return user_saved

■Request

http://localhost:8000/user/

{
  "username": "string",
  "password": "string",
  "email": "string",
  "full_name": "string"
}

■Response

{
  "username": "string",
  "email": "string",
  "full_name": "string"
}

###Return list of models

■Source

main.py


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

■Request

http://localhost:8000/items/

■Response

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

##Response Status Code

status_codeIt is possible to set the HTTP status code by using.

###Status code setting

■Source

main.py


from fastapi import FastAPI

app = FastAPI()

@app.post("/items/", status_code=201)
async def create_item(name: str):
    return {"name": name}
img5.jpg

###Status code shortcut

■Source

main.py


from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}

##Form Data

###Receiving form data

■Source

main.py


from fastapi import FastAPI, Form

app = FastAPI()

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

■Request

bash


curl -X POST "http://localhost:8000/login/" -H  "accept: application/json" -H  "Content-Type: application/x-www-form-urlencoded" -d "username=test&password=123"

##Request Files

###Send file

■Source

main.py


from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.post("/files/")
async def create_file(file: bytes = File(...)):
    return {"file_size": len(file)}

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"file": file}

@app.get("/")
async def main():
    content = """
    <html>
    <head>
      <title>FastAPI Form Test</title>
    </head>
    <body>
      <div>files</div>
      <form action="/files/" enctype="multipart/form-data" method="post">
        <input name="files" type="file">
        <input type="submit">
      </form>
      <div>uploadfiles</div>
      <form action="/uploadfiles/" enctype="multipart/form-data" method="post">
        <input name="files" type="file">
        <input type="submit">
      </form>
    </body>
    """
    return HTMLResponse(content=content)

■Request

bash


curl -X POST "http://localhost:8000/files/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "[email protected];type=image/jpeg"

■Response

{
  "file_size": 386446
}

■Request

bash


curl -X POST "http://localhost:8000/uploadfile/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "[email protected];type=image/jpeg"

■Response

{
  "file": {
    "filename": "img2.jpg ",
    "content_type": "image/jpeg",
    "file": {
      "_file": {},
      "_max_size": 1048576,
      "_rolled": false,
      "_TemporaryFileArgs": {
        "mode": "w+b",
        "buffering": -1,
        "suffix": null,
        "prefix": null,
        "encoding": null,
        "newline": null,
        "dir": null,
        "errors": null
      }
    }
  }
}

###Send multiple files

■Source

main.py


from typing import List
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.post("/files/")
async def create_files(files: List[bytes] = File(...)):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}

@app.get("/")
async def main():
    content = """
    <html>
    <head>
      <title>FastAPI Form Test</title>
    </head>
    <body>
      <div>files</div>
      <form action="/files/" enctype="multipart/form-data" method="post">
        <input name="files" type="file" multiple>
        <input type="submit">
      </form>
      <div>uploadfiles</div>
      <form action="/uploadfiles/" enctype="multipart/form-data" method="post">
        <input name="files" type="file" multiple>
        <input type="submit">
      </form>
    </body>
    """
    return HTMLResponse(content=content)

■Request

bash


curl -X POST "http://localhost:8000/files/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "[email protected];type=image/jpeg" -F "[email protected];type=image/jpeg"

■Response

{
  "file_sizes": [
    386446,
    320754
  ]
}

■Request

bash


curl -X POST "http://localhost:8000/uploadfiles/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "[email protected];type=image/jpeg" -F "[email protected];type=image/jpeg"

■Response

{
  "filenames": [
    "img1.jpg ",
    "img2.jpg "
  ]
}

##Request Forms and Files

###Submit form and file at the same time

■Source

main.py


from fastapi import FastAPI, File, Form, UploadFile

app = FastAPI()

@app.post("/files/")
async def create_file(
    file: bytes = File(...), fileb: UploadFile = File(...), token: str = Form(...)
):
    return {
        "file_size": len(file),
        "token": token,
        "fileb_content_type": fileb.content_type,
    }

■Request

bash


curl -X POST "http://localhost:8000/files/" -H  "accept: application/json" -H  "Content-Type: multipart/form-data" -F "token=aaa" -F "[email protected];type=image/jpeg" -F "[email protected];type=image/jpeg"

■Response

{
  "file_size": 386446,
  "token": "aaa",
  "fileb_content_type": "image/jpeg"
}

##Handling Errors

###Returns the specified error

HTTPExceptionIt is possible to return the status code and message by using.

■Source

main.py


from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

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

■Request

http://localhost:8000/items/foo

■Response

{
  "item": "The Foo Wrestlers"
}

■Request

http://localhost:8000/items/bar

■Response

{
  "detail": "Item not found"
}

###Add custom header

■Source

main.py


from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "There goes my error"},
        )
    return {"item": items[item_id]}

■Request

http://localhost:8000/items-header/bar

■Response

{
  "detail": "Item not found"
}

content-length: 27
content-type: application/json
date: Sun26 Jul 2020 02:47:20 GMT
server: uvicorn
x-error: There goes my error

###Custom exception handler

/unicorns/yoloWhen you requestraise UnicornExceptionWill be. @app.exception_handler(UnicornException)Therefore, the processing is actually performed here.

■Source

main.py


from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

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

app = FastAPI()

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )

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

■Request

http://localhost:8000/unicorns/yolo

■Response

{
  "message": "Oops! yolo did something. There goes a rainbow..."
}

content-length: 63
content-type: application/json
date: Sun26 Jul 2020 02:52:12 GMT
server: uvicorn

###Request body display at the time of verification error

It is possible to add the request body sent to Response when there is a verification error.

■Source

main.py


from fastapi import FastAPI, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )

class Item(BaseModel):
    title: str
    size: int

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

■Request

http://localhost:8000/items/

{
  "title": "towel",
  "size": "XL"
}

■Response

{
  "detail": [
    {
      "loc": [
        "body",
        "size"
      ],
      "msg": "value is not a valid integer",
      "type": "type_error.integer"
    }
  ],
  "body": {
    "title": "towel",
    "size": "XL"
  }
}

##Path Operation Configuration

###Specifying the response status code

status_codeIt is possible to specify the response status code with.

■Source

main.py


from typing import Optional, Set
from fastapi import FastAPI, status
from pydantic import BaseModel

app = FastAPI()

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

@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item
img6.jpg

###Reflection in OpenAPI document

Reflect tags, summaries, descriptions, response descriptions, and obsolete states.

■Source

main.py


from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

@app.get(
    "/items/",
    tags=["items"],
    summary="summary test",
    description="description test",
    response_description="response test"
)
async def read_items():
    return [{"name":"Foo", "price":42}]

@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(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

@app.get("/users/", tags=["users"], deprecated=True)
async def read_users():
    return [{"username": "johndoe"}]
img7.jpg img8.jpg

##JSON Compatible Encoder

###Convert to JSON compatible data type

For DB that can store only JSON compatible data type, for example, when storing datetime type, it is necessary to convert it to a character string. jsonable_encoderIt is possible to convert to JSON compatible data by using.

■Source

main.py


from datetime import datetime
from typing import Optional
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

fake_db = {}

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Optional[str] = None

app = FastAPI()

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    fake_db[id] = json_compatible_item_data

##Body - Updates

###PUT(Replace)Attention

When replacing the data in which the model is defined, if the Field in which the initial value is set is omitted, the initial value is automatically stored.

■Source

main.py


from typing import List, Optional
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()

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

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

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

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    update_item_encoded = jsonable_encoder(item)
    items[item_id] = update_item_encoded
    return update_item_encoded

■Request

http://localhost:8000/items/bar

{
    "name": "Barz",
    "price": 3,
    "description": "description"
}

■Response

{
  "name": "Barz",
  "description": "description",
  "price": 3,
  "tax": 10.5,
  "tags": []
}

###Partial update by PATCH

exclude_unsetBy using, the model is partially updated with the passed data.

■Source

main.py


from typing import List, Optional
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()

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

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

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

@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)
    items[item_id] = jsonable_encoder(updated_item)
    return updated_item

■Request

http://localhost:8000/items/bar

{
  "name": "Barz",
  "price": 100,
  "tags": [
    "test1", "test2"
  ]
}

■Response

{
  "name": "Barz",
  "description": "The bartenders",
  "price": 100,
  "tax": 20.2,
  "tags": [
    "test1",
    "test2"
  ]
}

##Dependencies

###Use of common parameters

Predefined parametersDependsUsed in.

■Source

main.py


from typing import Optional
from fastapi import Depends, FastAPI

app = FastAPI()

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

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

■Request

http://localhost:8000/items/?limit=100

■Response

{
  "q": null,
  "skip": 0,
  "limit": 100
}

###Common parameters(class)Use of

If the parameter type and dependency call are the same,DependsParameters can be omitted.

■Source

main.py


from typing import Optional
from fastapi import Depends, FastAPI

app = FastAPI()

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

class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
 async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
async def read_items(commons: CommonQueryParams = Depends()):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

###Dependency hierarchy

■Source

main.py


from typing import Optional
from fastapi import Cookie, Depends, FastAPI

app = FastAPI()

def query_extractor(q: Optional[str] = None):
    return q

def query_or_cookie_extractor(
    q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):
    if not q:
        return last_query
    return q

@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
    return {"q_or_cookie": query_or_default}

##CORS(Cross-Origin Resource Sharning)

###CORS settings

Store the URL list of permitted origins in an array. Set the URL list when defining the middleware.

■Source

main.py


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

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

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

##Bigger Applications - Multiple Files

###File division by module structure

Splitusers.py、items.pyPrepare to process each route with include_routerAssociate the route with. prefixURL prefix,tagsOpenAPI tags can be set at once with.

■Source

main.py


from fastapi import FastAPI
from routers import users, items

app = FastAPI()

app.include_router(users.router)

app.include_router(
    items.router,
    prefix="/items",
    tags=["items"]
)

routers/users.py


from fastapi import APIRouter

router = APIRouter()

users = {
  "user1": {"name": "user1", "email": "[email protected]", "age": 20},
  "user2": {"name": "user2", "email": "[email protected]", "age": 30},
  "user3": {"name": "user3", "email": "[email protected]", "age": 40},
}

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

@router.get("/users/me", tags=["users"])
async def get_user_me():
    return {"name": "the current user"}

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

routers/items.py


from fastapi import APIRouter

router = APIRouter()

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

@router.get("/")
async def get_items():
    return {"items": items}

@router.get("/{item_id}")
async def get_item(item_id: str):
    return {"item": items[item_id]}

##Backgroud Tasks

###Background tasks

Performs the next process without waiting for the executed process, such as sending an email or updating data that takes a long time.

■Source

main.py


from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

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):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

■Request

http://ubuntu18:8000/send-notification/test%40example.com?q=test%20test

■Response

{
  "message": "Message sent"
}

log.txt


found query: test test
message to [email protected]

##Metadata and Docs URLs

###Changed title, version and description of OpenAPI document

OpenAPI can be customized by setting the parameters when creating a FastAPI instance.

Parameters Description
title title
description Description
version version
openapi_tags tag
docs_url Swagger URL
redoc_url ReDoc URL
openapi_url OpenAPI URL

■Source

main.py


from fastapi import FastAPI

tags_metadata = [
    {
        "name": "users",
        "description": "Operations with users. The **login** logic is also here.",
    },
    {
        "name": "items",
        "description": "Manage items. So _fancy_ they have their own docs.",
        "externalDocs": {
            "description": "Items external docs",
            "url": "https://fastapi.tiangolo.com/",
        },
    },
]

app = FastAPI(
    title="My Super Project",
    description="This is a very fancy project, with auto docs for the API and everything",
    version="2.5.0",
    openapi_tags=tags_metadata,
    docs_url="/api/v1/docs",
    # docs_url=None,
    redoc_url="/api/v1/redoc",
    # redoc_url=None,
    openapi_url="/api/v1/openapi.json",
    # openapi_url=None,
)

@app.get("/users/", tags=["users"])
async def get_users():
    return [{"name": "Harry"}, {"name": "Ron"}]


@app.get("/items/", tags=["items"])
async def get_items():
    return [{"name": "wand"}, {"name": "flying broom"}]

http://localhost:8000/api/v1/docs http://localhost:8000/api/v1/redoc http://localhost:8000/api/v1/openapi.json

img9.png

##Static Files

###mount

By using mount, static files can be provided independently of FastAPI Routing. In the example below/staticYou can use static files placed in the static directory by accessing.

■Source

main.py


from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

http://localhost:8000/static/static_files.txt http://localhost:8000/static/img9.png

##Testing

###Split test file

■Source

src/main.py


from fastapi import FastAPI

app = FastAPI()

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

test_main.py


from fastapi.testclient import TestClient

from src.main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

Refer to the following site to import the sibling hierarchy module.PYTHONPATHRun the test with environment variables set https://rinoguchi.hatenablog.com/entry/2019/11/29/130224

bash


# Set environment variables
$ export PYTHONPATH="..:$PYTHONPATH"

# Run the test
$ pytest

img10.png

###Complex test

■Source

src/main.py


from typing import Optional
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: Optional[str] = None

@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: 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: 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=400, detail="Item already exists")
    fake_db[item.id] = item
    return item

tests/test_main.py


from fastapi.testclient import TestClient
from src.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_inexistent_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 == 400
    assert response.json() == {"detail": "Item already exists"}

Recommended Posts

How to use FastAPI ① Tutorial --User Guide
How to use FastAPI ② Advanced --User Guide
How to use FastAPI ③ OpenAPI
Cython Tutorial: How to Use shared_ptr
How to use xml.etree.ElementTree
How to use Python-shell
How to use tf.data
How to use virtualenv
How to use Seaboan
How to use image-match
How to use shogun
How to use Pandas 2
How to use Virtualenv
How to use numpy.vectorize
How to use pytest_report_header
How to use partial
How to use Bio.Phylo
How to use SymPy
How to use x-means
How to use WikiExtractor.py
How to use IPython
How to use virtualenv
How to use Matplotlib
How to use iptables
How to use numpy
How to use TokyoTechFes2015
How to use venv
How to use dictionary {}
How to use Pyenv
How to use list []
How to use python-kabusapi
How to use OptParse
How to use return
How to use dotenv
How to use pyenv-virtualenv
How to use Go.mod
How to use imutils
How to use import
How to use the NHK program guide API
How to use Qt Designer
How to use search sorted
[gensim] How to use Doc2Vec
python3: How to use bottle (2)
Understand how to use django-filter
How to use the generator
[Python] How to use list 1
How to use Python argparse
How to use IPython Notebook
How to use Pandas Rolling
[Note] How to use virtualenv
Linux user addition, how to use the useradd command
How to use redis-py Dictionaries
Python: How to use pydub
[Python] How to use checkio
[Go] How to use "... (3 periods)"
How to use Django's GeoIp2
[Python] How to use input ()
How to use the decorator
[Introduction] How to use open3d
How to use Python lambda
How to use Jupyter Notebook