[PYTHON] Verwendung von FastAPI ① Tutorial

Was geschrieben steht

--Memo bei Verwendung von FastAPI (Es kann falsch sein, weil es ein persönliches Memo ist ...)

Implementiert durch Befolgen des offiziellen Website-Dokuments (Tutorial - Benutzerhandbuch)

Referenz

FastAPI

Umgebung

Erstellen Sie eine Docker-Umgebung und überprüfen Sie den Vorgang

Docker FastAPI

Anlaufen

`bash`


# main.Wenn sich py im Stammverzeichnis befindet
$ uvicorn main:app --reload --host 0.0.0.0 --port 8000

# main.Wenn sich py nicht im Stammverzeichnis befindet
$ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

First Step

Interactive API Docs

Swagger

Alternative API docs

ReDoc

Aktivieren Sie Open API vor der automatischen Dokumentgenerierung

openapi.json

Einfache schnelle API-Funktionalität

`python.main.py`


from fastapi import FastAPI # ①

    app = FastAPI() # ②

    @app.get("/") # ③
        def read_root(): # ④
            return {"message": "Hello World"} # ⑤

① Importieren Sie die erforderlichen Pakete ② Erstellen Sie eine Fast API-Instanz ③ Arbeiten Sie mit Path Operation Decorator und geben Sie den Pfad an

Operation(Methode)	Erläuterung
POST	hinzufügen
GET	Erhalten
PUT	aktualisieren
DELETE	Löschen

④ Geben Sie die Funktion an, die beim Empfang einer Anforderung mit der Pfadoperationsfunktion aufgerufen werden soll

⑤ Geben Sie den Inhalt zurück (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"
}

Typdeklaration / Validierung

Gibt automatisch einen Fehler zurück, wenn ein Parameter übergeben wird, der nicht vom angegebenen Typ ist

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

Path Operation Decorator Reihenfolge

Wenn es eine URL gibt, die Sie vor anderen Pfadparametern priorisieren möchten, achten Sie auf die Reihenfolge. Wenn Sie später / users / me schreiben, hat der Pfadparameter Vorrang.

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

Vordefinierte Pfadparameter

Mit Enum können gültige Pfadparameter vordefiniert werden.

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

###Pfadparameter einschließlich Pfad

Im PfadparameterpathEs ist möglich, den Pass durch Beschreiben zu erhalten.

■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

Wenn ein Parameter übergeben wird, der kein Pfadparameter ist, wird er automatisch als Abfrageparameter bestimmt.

###Ursprünglicher Wert

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

###Möglichkeit

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

###Typdeklaration

Im booleschen Typ1、True、on、yesAuch angegebentrueWird beurteilt.

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

###Mehrere Pfade und Abfrageparameter

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

###Erforderliche Abfrageparameter

■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

###Anfordern der Körpergenerierung nach Datenmodell

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

###Body + Pfadparameter anfordern

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

###Anforderungshauptteil + Pfadparameter + Abfrageparameter

■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

Es ist möglich, Validierung und Metadaten für Abfrageparameter festzulegen.

###Überprüfung / optionale Parameter

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

###Überprüfung / erforderliche Parameter

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

###Überprüfung / regulärer Ausdruck

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

###Listen Sie die Abfrageparameter auf(Kein Standard)

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

###Listen Sie die Abfrageparameter auf(Standardmäßig)

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

###Metadateneinstellungen

Die festgelegten Metadaten werden automatisch im OpenAPI-Dokument angezeigt. deprecatedEs kann eindeutig festgestellt werden, dass die Verwendung nicht empfohlen wird. aliasDurch die Verwendung ist es möglich, den Variablennamen in der Funktion von dem zu trennen, wenn er als Pfadparameter empfangen wird.

■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

###Metadateneinstellungen

Es ist auch möglich, Metadaten im Pfadparameter festzulegen.

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

###Numerische Validierung

■Source

Vergleichsoperator	Erläuterung
ge	das ist alles
gt	groß
le	Weniger als
lt	klein

`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

###Kombination aus Pfadparameter, Abfrageparameter und Anforderungshauptteil

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

###Mehrere Anforderungskörper

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

###Explizite Angabe des Anforderungshauptteils

Wenn Sie in der Funktion einen Parameter angeben, der kein Pfadparameter ist, betrachtet FastAPI ihn als Abfrageparameter. Explizit, wenn Sie den Anforderungshauptteil anstelle des Abfrageparameters möchten= Body(...)Wird beschrieben.

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

###Mehrere Anforderungskörper und Abfrageparameter

Neben dem expliziten Anforderungshauptteil können auch Abfrageparameter angegeben werden.

■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

###Legen Sie Meta-Tags für jedes Feld im Anforderungshauptteil fest

Durch das Festlegen eines Meta-Tags im Schema wird es automatisch in der Open API wiedergegeben.

■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

###Liste im Feld "Anfragetext" festlegen

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

###Verschachteltes Modell

■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 der Untermodelle

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

###Tief verschachteltes Modell

■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 der Modelle

■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

###Festlegen des Schemabeispiels in 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

###Festlegen eines Schemabeispiels für jedes Feld des Modells

■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

###Festlegen eines Schemabeispiels in einer Funktion

■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

###Verwendung des Headers

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

###Doppelter Header

■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

###Stellen Sie ein anderes Ausgabemodell als die Eingabe ein

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

###Rückgabe des Standardwerts ausschließen

Normalerweise werden alle im Modell angegebenen Felder zurückgegeben. response_model_exclude_nusetNur das Feld, für das der Wert festgelegt ist, wird mit using zurückgegeben.

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

###Geben Sie das zurückzugebende Feld an

Es ist möglich, das vom Modell zurückzugebende Feld anzugeben.

■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

###Verwendung mehrerer Modelle

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

###Beseitigen Sie Modellduplikationen

Beseitigen Sie Doppelarbeit mithilfe der Modellvererbung.

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

###Liste der Modelle zurückgeben

■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_codeEs ist möglich, den HTTP-Statuscode mithilfe von festzulegen.

###Statuscode-Einstellung

■Source

`main.py`


from fastapi import FastAPI

app = FastAPI()

@app.post("/items/", status_code=201)
async def create_item(name: str):
    return {"name": name}

###Statuscode-Verknüpfung

■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

###Formulardaten empfangen

■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

###Datei senden

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

###Senden Sie mehrere Dateien

■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

###Formular und Datei gleichzeitig einreichen

■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

###Gibt den angegebenen Fehler zurück

HTTPExceptionEs ist möglich, den Statuscode und die Nachricht mithilfe von zurückzugeben.

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

###Benutzerdefinierten Header hinzufügen

■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

###Benutzerdefinierter Ausnahmebehandler

/unicorns/yoloWenn Sie anfordernraise UnicornExceptionWird sein. @app.exception_handler(UnicornException)Daher wird die Verarbeitung hier tatsächlich durchgeführt.

■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

###Anfordern der Körperanzeige zum Zeitpunkt des Überprüfungsfehlers

Es ist möglich, den an Response gesendeten Anforderungshauptteil hinzuzufügen, wenn ein Überprüfungsfehler vorliegt.

■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

###Angeben des Antwortstatuscodes

status_codeEs ist möglich, den Antwortstatuscode mit anzugeben.

■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

###Reflexion in der OpenAPI-Dokumentation

Reflektieren Sie Tags, Zusammenfassungen, Beschreibungen, Antwortbeschreibungen und veraltete Zustände.

■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

###In JSON-kompatiblen Datentyp konvertieren

Für eine Datenbank, in der nur JSON-kompatible Datentypen gespeichert werden können, z. B. beim Speichern des Datetime-Typs, muss dieser in eine Zeichenfolge konvertiert werden. jsonable_encoderEs ist möglich, mit JSON-kompatiblen Daten zu konvertieren.

■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(Ersetzen)Beachtung

Wenn beim Ersetzen der Daten, in denen das Modell definiert ist, das Feld weggelassen wird, in dem der Anfangswert festgelegt ist, wird der Anfangswert automatisch gespeichert.

■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": []
}

###Teilaktualisierung durch PATCH

exclude_unsetDurch die Verwendung wird das Modell teilweise mit den übergebenen Daten aktualisiert.

■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

###Verwendung gemeinsamer Parameter

Vordefinierte ParameterDependsBenutzt in.

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

###Gemeinsame Parameter(Klasse)Gebrauch von

Wenn der Parametertyp und der Abhängigkeitsaufruf identisch sindDependsParameter können weggelassen werden.

■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

###Abhängigkeitshierarchie

■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)

###CORS-Einstellungen

Speichert eine Liste der zulässigen Ursprungs-URLs in einem Array. Legen Sie die URL-Liste beim Definieren der Middleware fest.

■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

###Dateiteilung nach Modulstruktur

Teiltusers.py、items.pyBereiten Sie sich darauf vor, jede Route mit zu verarbeiten include_routerVerknüpfen Sie die Route mit. prefixURL-Präfix,tagsSie können OpenAPI-Tags sofort mit festlegen.

■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

###Hintergrundaufgaben

Führt den nächsten Prozess aus, ohne auf den ausgeführten Prozess zu warten, z. B. das Senden einer E-Mail oder das Aktualisieren von Daten, was lange dauert.

■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

###Titel, Version und Beschreibung der OpenAPI-Dokumentation wurden geändert

OpenAPI kann durch Festlegen von Parametern beim Erstellen einer FastAPI-Instanz angepasst werden.

Parameter	Erläuterung
title	Titel
description	Erläuterung
version	Ausführung
openapi_tags	Etikett
docs_url	Swagger URL
redoc_url	ReDoc-URL
openapi_url	Öffnen Sie die API-URL

■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

###montieren

Durch die Verwendung von mount können statische Dateien unabhängig vom schnellen API-Routing bereitgestellt werden. Im folgenden Beispiel/staticDurch den Zugriff können Sie statische Dateien verwenden, die sich im statischen Verzeichnis befinden.

■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

###Testdatei teilen

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

Informationen zum Importieren des Geschwisterhierarchiemoduls finden Sie auf der folgenden Website.PYTHONPATHLegen Sie Umgebungsvariablen fest und führen Sie den Test aus https://rinoguchi.hatenablog.com/entry/2019/11/29/130224

`bash`


# Umgebungsvariablen festlegen
$ export PYTHONPATH="..:$PYTHONPATH"

# Führen Sie den Test aus
$ pytest

###Komplexer Test

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

[PYTHON] Verwendung von FastAPI ① Tutorial - Benutzerhandbuch

Was geschrieben steht

Referenz

Umgebung

Anlaufen

bash

Aktivieren Sie Open API vor der automatischen Dokumentgenerierung

Einfache schnelle API-Funktionalität

python.main.py

main.py

Typdeklaration / Validierung

main.py

Path Operation Decorator Reihenfolge

main.py

Vordefinierte Pfadparameter

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

bash

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

bash

main.py

bash

bash

main.py

bash

bash

main.py

bash

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

main.py

`bash`

`python.main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`bash`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`bash`

`main.py`

`bash`

`bash`

`main.py`

`bash`

`bash`

`main.py`

`bash`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`main.py`

`routers/users.py`

`routers/items.py`

`main.py`

`log.txt`

`main.py`

`main.py`

`src/main.py`