--Memo when using FastAPI (It may be wrong because it is a personal memo ...) --Implemented by following the official website document (Advanced --User Guide)
Build a Docker environment and check the operation
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
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)
Additional Status Codes
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
■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"
}
■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 |
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"
}
]
■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>
■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
■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)
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
■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"
}
■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
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)
Response Cookies
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
■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
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
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
http://localhost:8000/subapp/docs
Event: startup - shutdown
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
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
■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"
}
■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