Comment utiliser FastAPI ② Advanced - 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 ...) --Mise en œuvre en apprenant du document officiel du site Web (Advanced - User Guide)

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

Path Operation Advanced Configuration

Exclure de l'API ouverte

En décrivant ʻinclude_in_schema = False`, l'affichage de l'Open API peut être exclu.

■Source

main.py


import uvicorn

from typing import Optional, Set
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []

@app.post("/items/", include_in_schema=False, response_model=Item, summary="Create an item")
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    \f
    :param item: User input.
    """
    return item

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

image.png

Additional Status Codes

Changer le code de statut à renvoyer

Vous pouvez modifier le code de statut en spécifiant status_code et content dans JSONResponse et en le renvoyant. Notez que le code d'état est modifié directement dans le code et ne sera pas reflété dans la documentation OpenAPI.

■Source

main.py


import uvicorn

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

app = FastAPI()

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

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

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

■Request

http://localhost:8000/items/test

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

■Response

status_code:201

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

Return a Response Directly

Renvoie une réponse directe

■Source

main.py


import uvicorn

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

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Optional[str] = None

app = FastAPI()

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    return JSONResponse(content=json_compatible_item_data)

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

■Request

http://localhost:8000/items/1

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

■Response

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

Renvoyer XML

■Source

main.py


import uvicorn

from fastapi import FastAPI, Response

app = FastAPI()

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

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

■Request

http://localhost:8000/legacy/

■Response

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

Custom Response - HTML, Stream, File, others

Il est possible de renvoyer des types de fichiers autres que le JSON par défaut.

Paramètres Type de données
content str or bytes
status_code int
headers dict
media_type str

Génération JSON orientée performances

Utilisez le package ʻorjson`.

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import ORJSONResponse

app = FastAPI()

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

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

■Request

http://localhost:8000/items/

■Response

[
  {
    "item_id": "Foo"
  }
]

Renvoyer le HTML

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()

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

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

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

■Request

http://localhost:8000/items/

■Response

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

Texte de retour

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import PlainTextResponse

app = FastAPI()

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

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

■Request

http://localhost:8000/

■Response

Hello World

Fichier

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import FileResponse

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

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

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

Spécifiez la classe de réponse par défaut

Vous pouvez spécifier la classe de réponse par défaut en utilisant default_response_class lors de la création d'une instance FastAPI.

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.responses import ORJSONResponse

app = FastAPI(default_response_class=ORJSONResponse)

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

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

Additional Responses in OpenAPI

Ajoutez votre propre modèle de réponse

■Source

main.py


import uvicorn

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

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

class Message(BaseModel):
    message: str

app = FastAPI()

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

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

■Request

http://localhost:8000/items/bar

■Response

status_code:404

{
  "message": "Item not found"
}

Renvoyer plusieurs types de fichiers du même domaine

■Source

main.py


import uvicorn

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

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

app = FastAPI()

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

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

■Request

http://localhost:8000/items/1

■Response

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

■Request

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

■Response

Image spécifiée

Prédéfini / combinaison du modèle de réponse d'origine

«302», «403»: définis à l'avance et déployés dans chaque domaine «200», «404»: définis individuellement pour chaque domaine

■Source

main.py


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

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

class Message(BaseModel):
    message: str

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

app = FastAPI()

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

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

image.png

Response Cookies

Renvoyer le cookie avec le paramètre de réponse

Des cookies peuvent être inclus dans la réponse. Le cookie renvoyé est stocké dans le navigateur

■Source

main.py



import uvicorn

from fastapi import FastAPI, Response

app = FastAPI()

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

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

Response Header

■Source

main.py


import uvicorn

from fastapi import FastAPI, Response

app = FastAPI()

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

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

■Request

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

■Response

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

Response - Change Status Code

Modifier le code d'état avec le paramètre Response

■Source

main.py


import uvicorn

from fastapi import FastAPI, Response, status

app = FastAPI()

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

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

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

■Request

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

■Response

"This didn't exist before"

Using the Request Directly

Accès Demande d'informations

Voir ci-dessous pour plus de détails sur l'objet Request. https://www.starlette.io/requests/

■Source

main.py


import uvicorn

from fastapi import FastAPI, Request

app = FastAPI()

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

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

■Request

http://localhost:8000/items/1

■Response

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

Sub Applications - Mounts

Application API Mount Sub Fast

Il est possible de définir des sous-applications sous une URL spécifique indépendamment de l'application principale Fast API.

■Source

main.py


import uvicorn

from fastapi import FastAPI

app = FastAPI()

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

subapp = FastAPI()

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

app.mount("/subapp", subapp)

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

http://localhost:8000/docs

image.png

http://localhost:8000/subapp/docs

image.png

Event: startup - shutdown

Événements lors du démarrage / arrêt de FastAPI

Avec ʻon_event`, vous pouvez définir le traitement lorsque FastAPI démarre et s'arrête.

■Source

main.py


import uvicorn

from fastapi import FastAPI

app = FastAPI()

items = {}

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

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

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

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

Custom Request and APIRoute class

tester

Si vous avez besoin d'exécuter startup et shutdown dans le test, utilisezwith TestClient ・ ・ ・.

■Source

main.py


import uvicorn

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

items = {}

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

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

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

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

Settings and Environment Variables

Utilisation de variables d'environnement

■Source

main.py


import uvicorn
from fastapi import FastAPI

import os

app = FastAPI()

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

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

■Request

Définir les variables d'environnement

bash


$ export MY_NAME="test"
$ echo $MY_NAME

http://ubuntu18:8000/

■Response

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

Utilisation du fichier .env

■Source

.env


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

config.py


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

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

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

# print(NAME)
# print(EMAIL)

main.py


import uvicorn

from fastapi import FastAPI
import config

app = FastAPI()

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

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

■Request

http://ubuntu18:8000/info

■Response

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

Recommended Posts

Comment utiliser FastAPI ② Advanced - Guide de l'utilisateur
Comment utiliser FastAPI ① Tutoriel - Guide de l'utilisateur
Comment utiliser FastAPI ③ OpenAPI
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 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
[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
[Python] Comment utiliser virtualenv
python3: Comment utiliser la bouteille (3)
python3: Comment utiliser la bouteille