How to use FastAPI ② Advanced --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 document (Advanced --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

Path Operation Advanced Configuration

Exclude from OpenAPI

By describing ʻinclude_in_schema = False`, the display of OpenAPI can be excluded.

■Source

main.py


import uvicorn

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.post("/items/", include_in_schema=False, response_model=Item, summary="Create an item")
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
    \f
    :param item: User input.
    """
    return item

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

image.png

Additional Status Codes

Change the status code to be returned

The status code can be changed by specifying status_code and content in JSONResponse and returning it. Note that the status code is changed directly in the code, so it will not be reflected in the OpenAPI documentation.

■Source

main.py


import uvicorn

from typing import Optional
from fastapi import Body, FastAPI, status
from fastapi.responses import JSONResponse

app = FastAPI()

items = {"foo": {"name": "Fighters", "size": 6}, "bar": {"name": "Tenders", "size": 3}}

@app.put("/items/{item_id}")
async def upsert_item(
    item_id: str, name: Optional[str] = Body(None), size: Optional[int] = Body(None)
):
    if item_id in items:
        item = items[item_id]
        item["name"] = name
        item["size"] = size
        return item
    else:
        item = {"name": name, "size": size}
        items[item_id] = item
        return JSONResponse(status_code=status.HTTP_201_CREATED, content=item)

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

■Request

http://localhost:8000/items/test

{
  "name": "test",
  "size": 100
}

■Response

status_code:201

{
  "name": "test",
  "size": 100
}

Return a Response Directly

Returns a direct response

■Source

main.py


import uvicorn

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

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)
    return JSONResponse(content=json_compatible_item_data)

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

■Request

http://localhost:8000/items/1

{
  "title": "string",
  "timestamp": "2020-07-28T04:58:58.605Z",
  "description": "string"
}

■Response

{
  "title": "string",
  "timestamp": "2020-07-28T04:58:58.605000+00:00",
  "description": "string"
}

Return XML

■Source

main.py


import uvicorn

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/legacy/")
def get_legacy_data():
    data = """<?xml version="1.0"?>
    <shampoo>
    <Header>
        Apply shampoo here.
    </Header>
    <Body>
        You'll have to use soap here.
    </Body>
    </shampoo>
    """
    return Response(content=data, media_type="application/xml")

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

■Request

http://localhost:8000/legacy/

■Response

<?xml version="1.0"?>
  <shampoo>
    <Header>
        Apply shampoo here.
    </Header>
    <Body>
        You'll have to use soap here.
    </Body>
  </shampoo>

Custom Response - HTML, Stream, File, others

It is possible to return file types other than the default JSON.

Parameters Data type
content str or bytes
status_code int
headers dict
media_type str

Performance-focused JSON generation

Use the ʻorjson` package.

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import ORJSONResponse

app = FastAPI()

@app.get("/items/", response_class=ORJSONResponse)
async def read_items():
    return [{"item_id": "Foo"}]

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

■Request

http://localhost:8000/items/

■Response

[
  {
    "item_id": "Foo"
  }
]

Return HTML

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()

def generate_html_response():
    html_content = """
<html>
  <head>
    <title>Some HTML in here</title>
  </head>
  <body>
    <h1>Look ma! HTML!</h1>
  </body>
</html>
    """
    return HTMLResponse(content=html_content, status_code=200)

@app.get("/items/", response_class=HTMLResponse)
async def read_items():
    return generate_html_response()

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

■Request

http://localhost:8000/items/

■Response

<html>
    <head>
        <title>Some HTML in here</title>
    </head>
    <body>
        <h1>Look ma! HTML!</h1>
    </body>
</html>

Return Text

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import PlainTextResponse

app = FastAPI()

@app.get("/", response_class=PlainTextResponse)
async def main():
    return "Hello World"

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

■Request

http://localhost:8000/

■Response

Hello World

File

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import FileResponse

some_file_path = "img/test.jpeg "
app = FastAPI()

@app.get("/")
async def main():
    return FileResponse(some_file_path)

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

Specify default response class

You can specify the default response class by using default_response_class when creating a FastAPI instance.

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import ORJSONResponse

app = FastAPI(default_response_class=ORJSONResponse)

@app.get("/items/")
async def read_items():
    return [{"item_id": "Foo"}]

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

Additional Responses in OpenAPI

Add your own response model

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel

class Item(BaseModel):
    id: str
    value: str

class Message(BaseModel):
    message: str

app = FastAPI()

@app.get("/items/{item_id}", response_model=Item, responses={404: {"model": Message}})
async def read_item(item_id: str):
    if item_id == "foo":
        return {"id": "foo", "value": "there goes my hero"}
    else:
        return JSONResponse(status_code=404, content={"message": "Item not found"})

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

■Request

http://localhost:8000/items/bar

■Response

status_code:404

{
  "message": "Item not found"
}

Return multiple file types from the same domain

■Source

main.py


import uvicorn

from typing import Optional
from fastapi import FastAPI
from fastapi.responses import FileResponse
from pydantic import BaseModel

class Item(BaseModel):
    id: str
    value: str

app = FastAPI()

@app.get(
    "/items/{item_id}",
    response_model=Item,
    responses={
        200: {
            "content": {"image/png": {}},
            "description": "Return the JSON item or an image.",
        }
    },
)
async def read_item(item_id: str, img: Optional[bool] = None):
    if img:
        return FileResponse("img/test.jpeg ", media_type="image/png")
    else:
        return {"id": "foo", "value": "there goes my hero"}

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

■Request

http://localhost:8000/items/1

■Response

{
  "id": "foo",
  "value": "there goes my hero"
}

■Request

http://localhost:8000/items/1?img=true

■Response

Specified image

Predefinition / combination of original response model

302, 403: Defined in advance and expanded in each domain 200, 404: Defined individually for each domain

■Source

main.py


from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel

class Item(BaseModel):
    id: str
    value: str

class Message(BaseModel):
    message: str

responses = {
    302: {"description": "The item was moved"},
    403: {"description": "Not enough privileges"},
}

app = FastAPI()

@app.get(
    "/items/{item_id}",
    response_model=Item,
    responses={
        **responses,
        404: {"model": Message, "description": "The item was not found"},
        200: {
            "description": "Item requested by ID",
            "content": {
                "application/json": {
                    "example": {"id": "bar", "value": "The bar tenders"}
                }
            },
        },
    },
)
async def read_item(item_id: str):
    if item_id == "foo":
        return {"id": "foo", "value": "there goes my hero"}
    else:
        return JSONResponse(status_code=404, content={"message": "Item not found"})

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

image.png

Response Cookies

Return cookie with Response parameter

Cookies can be included in the response. The returned cookie is stored in your browser

■Source

main.py



import uvicorn

from fastapi import FastAPI, Response

app = FastAPI()

@app.post("/cookie-and-object/")
def create_cookie(response: Response):
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    return {"message": "Come to the dark side, we have cookies"}

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

Response Header

■Source

main.py


import uvicorn

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/headers-and-object/")
def get_headers(response: Response):
    response.headers["X-Cat-Dog"] = "alone in the world"
    return {"message": "Hello World"}

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

■Request

http://localhost:8000/headers-and-object/

■Response

{
  "message": "Hello World"
}
content-length: 25 
content-type: application/json 
date: Tue28 Jul 2020 07:02:25 GMT 
server: uvicorn 
x-cat-dog: alone in the world 

Response - Change Status Code

Change status code with Response parameter

■Source

main.py


import uvicorn

from fastapi import FastAPI, Response, status

app = FastAPI()

tasks = {"foo": "Listen to the Bar Fighters"}

@app.put("/get-or-create-task/{task_id}", status_code=200)
def get_or_create_task(task_id: str, response: Response):
    if task_id not in tasks:
        tasks[task_id] = "This didn't exist before"
        response.status_code = status.HTTP_201_CREATED
    return tasks[task_id]

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

■Request

http://localhost:8000/get-or-create-task/bar

■Response

"This didn't exist before"

Using the Request Directly

Access Request information

See below for details on the Request object. https://www.starlette.io/requests/

■Source

main.py


import uvicorn

from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/items/{item_id}")
def read_root(item_id: str, request: Request):
    return {
        "item_id": item_id,
        "method": request.method,
        "url": request.url,
        "host": request.client.host,
        "port": request.client.port,
    }

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

■Request

http://localhost:8000/items/1

■Response

{
  "item_id": "1",
  "method": "GET",
  "url": {
    "_url": "http://localhost:8000/items/1"
  },
  "host": "172.24.0.1",
  "port": 51072
}

Sub Applications - Mounts

Mount Sub Fast API application

It is possible to define a sub-application under a specific URL independently of the main Fast API application.

■Source

main.py


import uvicorn

from fastapi import FastAPI

app = FastAPI()

@app.get("/items")
def read_main_item():
    return {
        "item_id": "main",
        "item_name": "main"
    }

subapp = FastAPI()

@subapp.get("/items")
def read_sub_item():
    return {
        "item_id": "sub",
        "item_name": "sub"
    }

app.mount("/subapp", subapp)

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

http://localhost:8000/docs

image.png

http://localhost:8000/subapp/docs

image.png

Event: startup - shutdown

Events when starting / stopping FastAPI

With ʻon_event`, you can set the process when FastAPI starts and stops.

■Source

main.py


import uvicorn

from fastapi import FastAPI

app = FastAPI()

items = {}

@app.on_event("startup")
async def startup_event():
    print("*** startup event ***")
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}

@app.on_event("shutdown")
def shutdown_event():
    print("*** shutdown event ***")
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")

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

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

Custom Request and APIRoute class

test

If you need to execute startup and shutdown in the test, use with TestClient ・ ・ ・.

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

items = {}

@app.on_event("startup")
async def startup_event():
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}

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

def test_read_items():
    with TestClient(app) as client:
        response = client.get("/items/foo")
        assert response.status_code == 200
        assert response.json() == {"name": "Fighters"}

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

Settings and Environment Variables

Use of environment variables

■Source

main.py


import uvicorn
from fastapi import FastAPI

import os

app = FastAPI()

@app.get("/")
def get_env():
    name = os.getenv("MY_NAME", "default")
    print(f"Hello {name} from Python")
    return {"name": f"Hello {name} from Python"}

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

■Request

Set environment variables

bash


$ export MY_NAME="test"
$ echo $MY_NAME

http://ubuntu18:8000/

■Response

{
  "name": "Hello test from Python"
}

Use of .env file

■Source

.env


NAME="AdminApp"
EMAIL="[email protected]"

config.py


import os
from os.path import join, dirname
from dotenv import load_dotenv

env_path = join(dirname(__file__), '.env')
load_dotenv(env_path)

NAME = os.environ.get("NAME")
EMAIL = os.environ.get("EMAIL")

# print(NAME)
# print(EMAIL)

main.py


import uvicorn

from fastapi import FastAPI
import config

app = FastAPI()

@app.get("/info")
async def info():
    return {
        "NAME": config.NAME,
        "EMAIL": config.EMAIL,
    }

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

■Request

http://ubuntu18:8000/info

■Response

{
  "NAME": "AdminApp",
  "EMAIL": "[email protected]"
}

Recommended Posts

How to use FastAPI ② Advanced --User Guide
How to use FastAPI ① Tutorial --User Guide
How to use FastAPI ③ OpenAPI
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 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
[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
[Python] How to use virtualenv
python3: How to use bottle (3)
python3: How to use bottle