[PYTHON] Comment utiliser FastAPI ① Tutoriel - Guide de l'utilisateur

Ce qui est écrit

--Mémo lors de l'utilisation de FastAPI (Cela peut être faux car il s'agit d'un mémo personnel ...)

référence

FastAPI

environnement

Créez un environnement Docker et vérifiez le fonctionnement

Docker FastAPI

Commencez

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

Swagger

Alternative API docs

ReDoc

Vérifiez l'Open API avant la génération automatique de document

openapi.json

Fonctionnalité API simple et rapide

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

http://localhost:8000/items/1

■Response

{
  "item_id": "1"
}

Déclaration / validation de type

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"
    }
  ]
}

Ordre du décorateur d'opération de chemin

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

Paramètres de chemin prédéfinis

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 cheminpathIl 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éen1TrueonyesÉgalement spécifiétrueEst 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. deprecatedIl peut être clairement indiqué qu'il n'est pas recommandé d'utiliser. aliasEn 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
img1.jpg img2.jpg

##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
img3.jpg

##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
img4.jpg

##Extra Data Types

■Source

main.py


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

app = FastAPI()

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

■Request

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

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

■Response

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

##Header Parameters

###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_nusetSeul 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_codeIl 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}
img5.jpg

###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

HTTPExceptionIl 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/yoloLorsque vous demandezraise UnicornExceptionSera. @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_codeIl 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
img6.jpg

###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"}]
img7.jpg img8.jpg

##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_encoderIl 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_unsetEn 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éfinisDependsUtilisé 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 identiquesDependsLes 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.pyitems.pyPréparez-vous à traiter chaque itinéraire avec include_routerAssociez l'itinéraire avec. prefixPréfixe d'URL,tagsVous 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

img9.png

##Static Files

###monter

En utilisant mount, les fichiers statiques peuvent être fournis indépendamment du routage API rapide. Dans l'exemple ci-dessous/staticVous 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.PYTHONPATHDé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

img10.png

###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

Comment utiliser FastAPI ① Tutoriel - Guide de l'utilisateur
Comment utiliser FastAPI ② Advanced - Guide de l'utilisateur
Comment utiliser FastAPI ③ OpenAPI
Tutoriel Cython: Comment utiliser shared_ptr
Comment utiliser xml.etree.ElementTree
Comment utiliser Python-shell
Remarques sur l'utilisation de tf.data
Comment utiliser virtualenv
Comment utiliser Seaboan
Comment utiliser la correspondance d'image
Comment utiliser le shogun
Comment utiliser Pandas 2
Comment utiliser Virtualenv
Comment utiliser numpy.vectorize
Comment utiliser pytest_report_header
Comment utiliser partiel
Comment utiliser Bio.Phylo
Comment utiliser SymPy
Comment utiliser x-means
Comment utiliser WikiExtractor.py
Comment utiliser IPython
Comment utiliser virtualenv
Comment utiliser Matplotlib
Comment utiliser iptables
Comment utiliser numpy
Comment utiliser TokyoTechFes2015
Comment utiliser venv
Comment utiliser le dictionnaire {}
Comment utiliser Pyenv
Comment utiliser la liste []
Comment utiliser python-kabusapi
Comment utiliser OptParse
Comment utiliser le retour
Comment utiliser pyenv-virtualenv
Comment utiliser imutils
Comment utiliser l'API du guide des programmes NHK
Comment utiliser Qt Designer
Comment utiliser la recherche triée
[gensim] Comment utiliser Doc2Vec
python3: Comment utiliser la bouteille (2)
Comprendre comment utiliser django-filter
Comment utiliser le générateur
[Python] Comment utiliser la liste 1
Comment utiliser Python Argparse
Comment utiliser IPython Notebook
Comment utiliser Pandas Rolling
[Note] Comment utiliser virtualenv
Ajouter un utilisateur Linux, comment utiliser la commande useradd
Comment utiliser les dictionnaires redis-py
Python: comment utiliser pydub
[Python] Comment utiliser checkio
[Aller] Comment utiliser "... (3 périodes)"
Comment faire fonctionner GeoIp2 de Django
[Python] Comment utiliser input ()
Comment utiliser le décorateur
[Introduction] Comment utiliser open3d
Comment utiliser Python lambda
Comment utiliser Jupyter Notebook