--Memo bei Verwendung von FastAPI (Es kann falsch sein, weil es ein persönliches Memo ist ...)
Erstellen Sie eine Docker-Umgebung und überprüfen Sie den Vorgang
bash
# main.Wenn sich py im Stammverzeichnis befindet
$ uvicorn main:app --reload --host 0.0.0.0 --port 8000
# main.Wenn sich py nicht im Stammverzeichnis befindet
$ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
Path Operation Advanced Configuration
Durch die Beschreibung von "include_in_schema = False" kann die Anzeige der Open API ausgeschlossen werden.
■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
Sie können den Statuscode ändern, indem Sie "Statuscode" und "Inhalt" in "JSONResponse" angeben und zurückgeben. Beachten Sie, dass der Statuscode direkt im Code geändert wird und nicht in der OpenAPI-Dokumentation berücksichtigt wird.
■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
Es ist möglich, andere Dateitypen als den Standard-JSON zurückzugeben.
Parameter | Datentyp |
---|---|
content | str or bytes |
status_code | int |
headers | dict |
media_type | str |
Verwenden Sie das Orjson-Paket.
■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)
Sie können die Standardantwortklasse angeben, indem Sie beim Erstellen einer FastAPI-Instanz "default_response_class" verwenden.
■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
Angegebenes Bild
302
, 403
: Im Voraus definiert und in jeder Domäne bereitgestellt
200
, 404
: Für jede Domain einzeln definiert
■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 können in die Antwort aufgenommen werden. Das zurückgegebene Cookie wird im Browser gespeichert
■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
Einzelheiten zum Anforderungsobjekt finden Sie weiter unten. 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
Es ist möglich, Unteranwendungen unter einer bestimmten URL unabhängig von der Hauptanwendung von Fast API zu definieren.
■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
Mit on_event
können Sie die Verarbeitung beim Starten / Stoppen der Fast API einstellen.
■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
Wenn Sie im Test "Starten" und "Herunterfahren" ausführen müssen, verwenden Sie "mit 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
Umgebungsvariablen festlegen
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