--Mémo lors de l'utilisation de FastAPI (Cela peut être faux car il s'agit d'un mémo personnel ...)
Créez un environnement Docker et vérifiez le fonctionnement
bash
# main.Si py est dans le répertoire racine
$ uvicorn main:app --reload --host 0.0.0.0 --port 8000
# main.Si py n'est pas dans le répertoire racine
$ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
First Step
Interactive API Docs
Alternative API docs
python.main.py
from fastapi import FastAPI # ①
app = FastAPI() # ②
@app.get("/") # ③
def read_root(): # ④
return {"message": "Hello World"} # ⑤
① Importez les packages requis ② Créer une instance d'API Fast ③ Opérez avec Path Operation Decorator et spécifiez le chemin
opération(Méthode) | La description |
---|---|
POST | ajouter à |
GET | Avoir |
PUT | mise à jour |
DELETE | Effacer |
④ Spécifiez la fonction à appeler lors de la réception d'une requête avec Path Operation Function
⑤ Renvoyer le contenu (JSON)
Path Parameters
Basic
■
main.py
@app.get("/items/{item_id}")
async def get_item(item_id):
return { "item_id": item_id }
■Request
■Response
{
"item_id": "1"
}
Renvoie automatiquement une erreur si un paramètre qui n'est pas du type spécifié est passé
■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"
}
]
}
S'il existe une URL que vous souhaitez privilégier sur les autres paramètres de chemin, faites attention à l'ordre. Si vous écrivez / users / me plus tard, le paramètre de chemin sera prioritaire.
■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"
}
Il est possible de prédéfinir des paramètres de chemin valides en utilisant 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"
]
}
}
]
}
###Paramètres du chemin, y compris le chemin
Dans le paramètre de cheminpath
Il est possible de recevoir le pass en décrivant.
■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
Si un paramètre qui n'est pas un paramètre de chemin est passé, il est automatiquement déterminé en tant que paramètre de requête.
###valeur initiale
■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"
}
###Déclaration de type
En type booléen1
、True
、on
、yes
Également spécifiétrue
Est jugé.
###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"
}
###Plusieurs chemins et paramètres de requête
■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"
}
###Paramètres de requête obligatoires
■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
###Demander la génération de corps par modèle de données
■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
}
###Demander les paramètres du corps et du chemin
■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
}
###Corps de la requête + paramètre de chemin + paramètre de requête
■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
Il est possible de définir la validation et les métadonnées pour les paramètres de requête.
###Vérification / paramètres facultatifs
■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
}
}
]
}
###Vérification / paramètres requis
■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"
}
###Vérification / expression régulière
■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"
}
}
]
}
###Liste des paramètres de requête(Aucun défaut)
■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"
]
}
###Liste des paramètres de requête(Par défaut)
■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"
]
}
###Paramètres de métadonnées
Les métadonnées définies sont automatiquement reflétées dans le document OpenAPI.
deprecated
Il peut être clairement indiqué qu'il n'est pas recommandé d'utiliser.
alias
En utilisant, il est possible de séparer le nom de la variable dans la fonction de lors de sa réception en tant que paramètre de chemin.
■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
##Path Parameters and Numeric Validations
###Paramètres de métadonnées
Il est également possible de définir des métadonnées dans le paramètre de chemin.
■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
}
###Validation numérique
■Source
Opérateur de comparaison | La description |
---|---|
ge | c'est tout |
gt | grand |
le | Moins que |
lt | petit |
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
###Combinaison de paramètre de chemin, paramètre de requête, corps de requête
■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
}
}
###Plusieurs corps de demande
■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"
}
}
###Spécification explicite du corps de la requête
Si vous spécifiez un paramètre dans la fonction qui n'est pas un paramètre de chemin, FastAPI le considère comme un paramètre de requête.
Explicitement si vous souhaitez que le corps de la requête au lieu du paramètre de requête= Body(...)
Est décrit.
■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
}
###Plusieurs corps de requête et paramètres de requête
En plus du corps de la requête explicite, des paramètres de requête peuvent également être spécifiés.
■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
###Définir des balises meta pour chaque champ dans le corps de la demande
En définissant une balise Meta dans le schéma, elle est automatiquement reflétée dans l'Open API.
■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
##Body - Nested Models
###Définir la liste dans le champ du corps de la demande
■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"
]
}
}
###Modèle imbriqué
■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"
}
}
}
###Liste des sous-modèles
■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"
}
]
}
}
###Modèle profondément imbriqué
■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"
}
]
}
]
}
###Liste des modèles
■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
###Définition de l'exemple de schéma dans 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
###Définition d'un exemple de schéma pour chaque champ du modèle
■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
###Définition d'un exemple de schéma dans une fonction
■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
##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
###Utilisation de l'en-tête
■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"
}
###En-tête en double
■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
###Définir un modèle de sortie différent de l'entrée
■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"
}
###Exclure le retour de la valeur par défaut
Normalement, tous les champs spécifiés dans le modèle sont renvoyés.
response_model_exclude_nuset
Seul le champ pour lequel la valeur est définie est renvoyé à l'aide de.
■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
}
###Spécifiez le champ à renvoyer
Il est possible de spécifier le champ à retourner du modèle.
■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
###Utilisation de plusieurs modèles
■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"
}
###Élimine la duplication des modèles
Éliminez la duplication en utilisant l'héritage de modèle.
■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"
}
###Renvoyer la liste des modèles
■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_code
Il est possible de définir le code d'état HTTP en utilisant.
###Réglage du code d'état
■Source
main.py
from fastapi import FastAPI
app = FastAPI()
@app.post("/items/", status_code=201)
async def create_item(name: str):
return {"name": name}
###Raccourci du code d'état
■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
###Réception des données du formulaire
■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
###Envoyer le fichier
■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
}
}
}
}
###Envoyer plusieurs fichiers
■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
###Soumettez le formulaire et déposez en même temps
■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
###Renvoie l'erreur spécifiée
HTTPException
Il est possible de renvoyer le code d'état et le message en utilisant.
■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"
}
###Ajouter un en-tête personnalisé
■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
###Gestionnaire d'exceptions personnalisé
/unicorns/yolo
Lorsque vous demandezraise UnicornException
Sera.
@app.exception_handler(UnicornException)
Par conséquent, le traitement est réellement effectué ici.
■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
###Demander l'affichage du corps au moment de l'erreur de vérification
Il est possible d'ajouter le corps de la requête envoyée à Response en cas d'erreur de vérification.
■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
###Spécification du code d'état de la réponse
status_code
Il est possible de spécifier le code d'état de la réponse avec.
■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
###Réflexion dans la documentation OpenAPI
Refléter les balises, les résumés, les descriptions, les descriptions de réponse et les états obsolètes.
■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"}]
##JSON Compatible Encoder
###Convertir en type de données compatible JSON
Pour une base de données qui ne peut stocker que des types de données compatibles JSON, par exemple, lors du stockage du type datetime, il est nécessaire de le convertir en chaîne de caractères.
jsonable_encoder
Il est possible de convertir en données compatibles JSON en utilisant.
■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(Remplacer)Attention
Si le champ dans lequel la valeur initiale est définie est omis lors du remplacement des données dans lesquelles le modèle est défini, la valeur initiale est automatiquement stockée.
■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": []
}
###Mise à jour partielle par PATCH
exclude_unset
En utilisant, le modèle est partiellement mis à jour avec les données transmises.
■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
###Utilisation de paramètres communs
Paramètres prédéfinisDepends
Utilisé dans.
■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
}
###Paramètres communs(classe)Utilisation de
Si le type de paramètre et l'appel de dépendance sont identiquesDepends
Les paramètres peuvent être omis.
■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
###Hiérarchie des dépendances
■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)
###Paramètres CORS
Stocke une liste des URL d'origine autorisées dans un tableau. Définissez la liste des URL lors de la définition du 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
###Division des fichiers par structure de module
Diviséusers.py
、items.py
Préparez-vous à traiter chaque itinéraire avec
include_router
Associez l'itinéraire avec.
prefix
Préfixe d'URL,tags
Vous pouvez définir des balises OpenAPI à la fois avec.
■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
###Tâches d'arrière-plan
Effectue le processus suivant sans attendre le processus exécuté, tel que l'envoi d'un e-mail ou la mise à jour des données qui prend beaucoup de temps.
■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
###Changement du titre, de la version et de la description de la documentation OpenAPI
OpenAPI peut être personnalisé en définissant des paramètres lors de la création d'une instance FastAPI.
Paramètres | La description |
---|---|
title | Titre |
description | La description |
version | version |
openapi_tags | marque |
docs_url | URL Swagger |
redoc_url | URL ReDoc |
openapi_url | Ouvrir l'URL de l'API |
■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
##Static Files
###monter
En utilisant mount, les fichiers statiques peuvent être fournis indépendamment du routage API rapide.
Dans l'exemple ci-dessous/static
Vous pouvez utiliser des fichiers statiques placés dans le répertoire statique en accédant à.
■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
###Fichier de test fractionné
■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"}
Reportez-vous au site suivant pour importer le module de hiérarchie des frères.PYTHONPATH
Définir les variables d'environnement et exécuter le test
https://rinoguchi.hatenablog.com/entry/2019/11/29/130224
bash
# Définir les variables d'environnement
$ export PYTHONPATH="..:$PYTHONPATH"
# Lancer le test
$ pytest
###Test complexe
■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