Request validation error fastapi

FastAPI has a great Exception Handling, so you can customize your exceptions in many ways.

You can raise an HTTPException, HTTPException is a normal Python exception with additional data relevant for APIs. But you can’t return it you need to raise it because it’s a Python exception

from fastapi import HTTPException
...
@app.get("/")
async def hello(name: str):
    if not name:
        raise HTTPException(status_code=404, detail="Name field is required")
    return {"Hello": name}

By adding name: str as a query parameter it automatically becomes required so you need to add Optional

from typing import Optional
...
@app.get("/")
async def hello(name: Optional[str] = None):
    error = {"Error": "Name field is required"}
    if name:
        return {"Hello": name}
    return error

$ curl 127.0.0.1:8000/?name=imbolc
{"Hello":"imbolc"}
...
$ curl 127.0.0.1:8000
{"Error":"Name field is required"}

But in your case, and i think this is the best way to handling errors in FastAPI overriding the validation_exception_handler:

from fastapi import FastAPI, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
...
@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(), "Error": "Name field is missing"}),
    )
...
@app.get("/")
async def hello(name: str):
    return {"hello": name}

You will get a response like this:

$ curl 127.0.0.1:8000

 {
   "detail":[
      {
         "loc":[
            "query",
            "name"
         ],
         "msg":"field required",
         "type":"value_error.missing"
      }
   ],
   "Error":"Name field is missing"
}

You can customize your content however if you like:

{
"Error":"Name field is missing",
   "Customize":{
      "This":"content",
      "Also you can":"make it simpler"
   }
}

Есть много ситуаций, когда вам нужно уведомить об ошибке клиента, использующего ваш API.

Этот клиент может быть браузером с внешним интерфейсом, чужим кодом, IoT-устройством и т. д.

Возможно, вам придется сообщить клиенту, что:

• У клиента недостаточно прав для этой операции.
• У клиента нет доступа к этому ресурсу.
• Элемент, к которому пытался получить доступ клиент, не существует.
• etc.

В этих случаях вы обычно возвращаете код состояния HTTP в диапазоне 400 (от 400 до 499).

Это похоже на 200 кодов состояния HTTP (от 200 до 299). Эти коды состояния «200» означают, что запрос каким-то образом был «успешен».

Коды состояния в диапазоне 400 означают, что произошла ошибка клиента.

Помните все эти ошибки «404 Not Found» (и шутки)?

Use HTTPException

Чтобы вернуть клиенту HTTP-ответы с ошибками, вы используете HTTPException .

Import HTTPException

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

Поднимите HTTPException в своем коде

HTTPException — это обычное исключение Python с дополнительными данными, относящимися к API.

Поскольку это исключение Python, вы его не return raise .

Это также означает, что если вы находитесь внутри служебной функции, которую вы вызываете внутри своей функции операции пути , и вы вызываете HTTPException изнутри этой служебной функции, она не будет запускать остальную часть кода в функции операции пути , он немедленно завершит этот запрос и отправит клиенту ошибку HTTP из исключения HTTPException .

Преимущество возбуждения исключения по сравнению с return значения будет более очевидным в разделе, посвященном зависимостям и безопасности.

В этом примере, когда клиент запрашивает элемент по несуществующему идентификатору, создайте исключение с кодом состояния 404 :

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

Полученная реакция

Если клиент запрашивает http://example.com/items/foo ( item_id "foo" ), этот клиент получит код состояния HTTP 200 и ответ JSON:

{
  "item": "The Foo Wrestlers"
}

Но если клиент запрашивает http://example.com/items/bar (несуществующий item_id "bar" ), этот клиент получит код состояния HTTP 404 (ошибка «не найден») и ответ JSON. из:

{
  "detail": "Item not found"
}

В некоторых ситуациях полезно иметь возможность добавлять собственные заголовки к ошибке HTTP. Например, для некоторых видов безопасности.

Вам, вероятно, не нужно будет использовать его непосредственно в коде.

Но если вам это нужно для расширенного сценария, вы можете добавить собственные заголовки:

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

Установите пользовательские обработчики исключений

Вы можете добавить собственные обработчики исключений теми же утилитами исключений от Starlette .

Допустим, у вас есть пользовательское исключение UnicornException , которое вы (или используемая вами библиотека) можете raise .

И вы хотите обрабатывать это исключение глобально с помощью FastAPI.

Вы можете добавить собственный обработчик исключений с помощью @app.exception_handler() :

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}

UnicornException вы запросите /unicorns/yolo , операция пути вызовет raise UnicornException .

Но это будет обрабатываться unicorn_exception_handler .

Итак, вы получите чистую ошибку с кодом состояния HTTP 418 и содержимым JSON:

{"message": "Oops! yolo did something. There goes a rainbow..."}

Переопределить обработчики исключений по умолчанию

FastAPI имеет несколько обработчиков исключений по умолчанию.

Эти обработчики отвечают за возврат ответов JSON по умолчанию, когда вы HTTPException raise когда запрос содержит недопустимые данные.

Вы можете переопределить эти обработчики исключений своими собственными.

Переопределить исключения проверки запроса

Когда запрос содержит недопустимые данные, FastAPI внутренне вызывает RequestValidationError .

И он также включает в себя обработчик исключений по умолчанию.

Чтобы переопределить его, импортируйте RequestValidationError и используйте его с @app.exception_handler(RequestValidationError) для оформления обработчика исключений.

Обработчик исключений получит Request и исключение.

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return PlainTextResponse(str(exc), status_code=400)


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

Теперь, если вы перейдете к /items/foo , вместо того, чтобы получить ошибку JSON по умолчанию с:

{
    "detail": [
        {
            "loc": [
                "path",
                "item_id"
            ],
            "msg": "value is not a valid integer",
            "type": "type_error.integer"
        }
    ]
}

вы получите текстовую версию с:

1 validation error
path -> item_id
  value is not a valid integer (type=type_error.integer)

RequestValidationError против ValidationError

RequestValidationError — это подкласс ValidationError Pydantic .

FastAPI использует его, поэтому, если вы используете модель Pydantic в response_model и в ваших данных есть ошибка, вы увидите ошибку в своем журнале.

Но клиент/пользователь этого не увидит. Вместо этого клиент получит «Внутреннюю ошибку сервера» с кодом состояния HTTP 500 .

Так и должно быть, потому что если у вас есть Pydantic ValidationError в вашем ответе или где-либо в вашем коде (не в запросе клиента ), это на самом деле ошибка в вашем коде.

И пока вы ее исправляете, ваши клиенты/пользователи не должны иметь доступа к внутренней информации об ошибке, так как это может привести к уязвимости системы безопасности.

Переопределить обработчик ошибок HTTPException

Точно так же вы можете переопределить обработчик HTTPException .

Например, вы можете захотеть вернуть простой текстовый ответ вместо JSON для этих ошибок:

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return PlainTextResponse(str(exc), status_code=400)


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

Используйте тело RequestValidationError

Ошибка RequestValidationError содержит полученное body

Вы можете использовать его при разработке своего приложения для регистрации тела и его отладки, возврата пользователю и т. д.

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

Теперь попробуйте отправить недопустимый элемент, например:

{
  "title": "towel",
  "size": "XL"
}

Вы получите ответ о том, что данные, содержащие полученное тело, недействительны:

{
  "detail": [
    {
      "loc": [
        "body",
        "size"
      ],
      "msg": "value is not a valid integer",
      "type": "type_error.integer"
    }
  ],
  "body": {
    "title": "towel",
    "size": "XL"
  }
}

HTTPException FastAPI против HTTPException Starlette

FastAPI имеет собственное HTTPException .

И класс HTTPException FastAPI HTTPException наследуется от класса ошибок HTTPException Starlette .

Единственное отличие состоит в том, что HTTPException FastAPI позволяет вам добавлять заголовки, которые будут включены в ответ.

Это необходимо/используется внутри OAuth 2.0 и некоторых утилит безопасности.

Таким образом, вы можете продолжать вызывать HTTPException FastAPI , HTTPException обычно, в своем коде.

Но когда вы регистрируете обработчик исключений, вы должны зарегистрировать его для HTTPException Starlette .

Таким образом, если какая-либо часть внутреннего кода Starlette, расширение или подключаемый модуль Starlette вызывает исключение HTTPException Starlette , ваш обработчик сможет его перехватить и обработать.

В этом примере, чтобы иметь возможность иметь оба HTTPException в одном и том же коде, исключения Starlette переименовываются в StarletteHTTPException :

from starlette.exceptions import HTTPException as StarletteHTTPException

Повторно использовать обработчики исключений FastAPI

Если вы хотите использовать исключение вместе с теми же обработчиками исключений по умолчанию из FastAPI , вы можете импортировать и повторно использовать обработчики исключений по умолчанию из fastapi.exception_handlers :

from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
    print(f"OMG! An HTTP error!: {repr(exc)}")
    return await http_exception_handler(request, exc)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    print(f"OMG! The client sent invalid data!: {exc}")
    return await request_validation_exception_handler(request, exc)


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

В этом примере вы просто print ошибку с очень выразительным сообщением, но вы поняли идею. Вы можете использовать исключение, а затем просто повторно использовать обработчики исключений по умолчанию.

FastAPI

0.86

• Extra Models

• First Steps

• Header Parameters

• Tutorial — User Guide — Intro

Install

pip install fastapi

You will need an ASGI server, the production environment can be usedUvicorn or Hypercorn。

pip install uvicorn

The simplest program

from fastapi import FastAPI

app = FastAPI()


 @ app.get ("/") // app.get refers to the GET request, can also be app.post, app.put, app.delete, etc.
async def root():
    return {"message": "Hello World"}

Run real-time server

uvicorn main:app --reload --port 5000 --host 0.0.0.0
 # Parameter information 
 Main refers to the program entry file name
 App is an instance name app = fastapi ()
 --Reload refers to automatic restart after modification

You can also use the following way to run, convenient for debugging

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Interactive API document

Visithttp://127.0.0.1:8000/docs。

You will see the automatically generated interactive API document (bySwagger UI supply):

Optional API document

Visithttp://127.0.0.1:8000/redoc。

You will see an optional automatic generation document (byReDoc supply):

Query parameters and string checks

Simple query parameters

The parameters on the URL will be passed as a function of the function.

from fastapi import FastAPI

app = FastAPI()


@app.get("/test")
 Async Def Test (Skip: INT = 0, LIMIT: INT = 3): # Skip default value of 0, Limit default is 3
    print(skip, limit)
    return skip + limit

Access http://127.0.0.1:8000/Items/?skip=0&limit=10, the page will display 10.

Access http: // localhost: 8000 / test? Limit = 20, page display 20, Skip default is 0, limit is URL incoming 20

The query parameter is:

• skip: The corresponding value is0
• limit: The corresponding value is10

Since they are part of the URL, their «original value» is a string.

However, when you declare the Python type (in the above exampleintWhen they convert them to this type and verify this type.

Optional parameters

In the same way, you can set their default values ​​toNone To declare an optional query parameter:

Typing.optional optional type,Equivalent with parameters with default valuesDifferent, using optional will tell you the IDE or framework: This parameter can be None except for a given default value, and use some static check tools such as MyPy, a similar statement like A: int = none may Tip report error, but use A: optional [int] = None will not.

from typing import Optional
from fastapi import FastAPI

app = FastAPI()


@app.get("/test/{id}")
 Async Def Test (ID: INT, Q: Optional [str] = none): # q is not required
    if q:
        return {"id": id, "q": q}
    return {"id": id}

In this example, the function parametersq Will be optional, and the default isNone。

Query parameter type conversion

from typing import Optional
from fastapi import FastAPI

app = FastAPI()


@app.get("/test/{id}")
async def test(id: int, q: Optional[str] = None, short: bool = False):
    print(type(short))  # <class 'bool'>
    if q:
        return {"id": id, "q": q}
    return {"id": id}

In this example, if you visit:

http://127.0.0.1:8000/items/foo?short=1

or

http://127.0.0.1:8000/items/foo?short=True

or

http://127.0.0.1:8000/items/foo?short=true

or

http://127.0.0.1:8000/items/foo?short=on

or

http://127.0.0.1:8000/items/foo?short=yes

Or any other variant form (uppercase, first letters, etc.), your function is receivedshort The parameters will be BooleanTrue. For the valueFalse The situation is true.

Multiple paths and query parameters

You can declare multiple path parameters and query parameters at the same time.FastAPI Can recognize them through the name.

from typing import Optional

from fastapi import FastAPI

app = FastAPI()


@app.get("/users/{user_id}/items/{item_id}")
async def read_user_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

Required query parameters

When you declare the default value for the non-path parameters (currently, there is only the query parameters we know), the parameter is not required.

If you don’t want to add a specific value, just want this parameter to be optional, set the default value toNone。

But when you want a query parameter to become required, you do not declare any default value:

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):
    item = {"item_id": item_id, "needy": needy}
    return item

The query parameters hereneedy Be the type ofstr Required query parameters.

When accessing http: // localhost: 8000 / items / 1, there is no Needy parameter and cannot be accessed.

Access http: // localhost: 8000 / items / 1? Needy = Needy can be accessed.

Use Query parameter to verify

We intend to add constraints: even ifq It is optional, but as long as this parameter is provided, the parameter value is provided.Can’t exceed 5 characters。

from typing import Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, max_length=5)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

use Query As a default

NowQuery Used as the default value of query parameters, and put itmax_length The parameter is set to 5

from typing import Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Optional[str] = Query(123, max_length=5)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

Because we must useQuery(None) Replace the default valueNone，Query The first parameter is also used to define default values.

so:

q: str = Query(None)

… make the parameters optional, equivalent:

q: str = None

but Query Explicitly declare it as query parameters.

Then we can pass more parameters toQuery. In this example, it is suitable for stringmax_length parameter:

q: str = Query(None, max_length=5)

Add regular expression check

You can define a regular expression that must be matched by a parameter value:

from typing import Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
 Async defread_items (q: optional [str] = query (123, max_length = 5, regex = r " d +")): # Using the regular display parameter is an integer
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

Query declares as required parameters

When we don’t need to declare additional check or metadata, you can make it only if you don’t declare the default value.q Parameters become required parameters, such as:

q: str

replace:

q: str = None

But now we are usingQuery Declare it, for example:

q: str = Query(None, min_length=3)

So when you are usingQuery And when you need to declare a value is required, you can... Used as the first parameter value:

from typing import Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Optional[str] = Query(..., max_length=5, regex=r"d+")):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

Query query parameter list / multiple values

When you useQuery When the query parameter is explicitly defined, you can also declare it to receive a set of values, or in other words, receive multiple values.

For example, to declare a multi-time query parameter in the URLqYou can write this:

from typing import List, Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Optional[List[str]] = Query(None)):
    query_items = {"q": q}
    return query_items  # {"q":["123","abc"]}

Enter the URL:http://localhost:8000/items/?q=123&q=abc

Response is: {«Q»: [«123», «ABC»]}

Query parameter list with default values

from typing import List, Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Optional[List[str]] = Query(["lowell", "xiaoqi"])):
    query_items = {"q": q}
    return query_items  # {"q":["lowell","xiaoqi"]}

use list

You can also use it directlylist replace List [str]：

from typing import List, Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: list = Query(["lowell", "xiaoqi"])):
    query_items = {"q": q}
    return query_items  # {"q":["lowell","xiaoqi"]}

Notice:In this case FASTAPI will not check the contents of the list.

E.g,List[int] The content of the list must be an integer must be an integer. But separatelist Will not.

Query More Data

You can see when using the interface documentation

from typing import List, Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
 Async Def Read_Items (Q: List = Query (["Lowell", "XIAOQI"], Title = "Name List", Description = "Recommended the name of the class"):
    query_items = {"q": q}
    return query_items  # {"q":["lowell","xiaoqi"]}

Query’s unique name parameters

Suppose you want to query parametersname

Like this:

http://localhost:8000/items/?name=123&name=456

But if you use Q in Q

At this time you can usealias The parameter declares an alias that will be used to find the query parameter value in the URL:

from typing import List, Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: list = Query(None, alias="name")):
    query_items = {"q": q}
    return query_items  # {"q":["lowell","xiaoqi"]}

Visit http: // localhost: 8000 / items /? Name = 123 & name = 456

Abandonment parameters

Now suppose you no longer like this parameter.

You have to keep it for a while, because some clients are using it, but you want the documentation to show it as a deprecated.

Then put the parametersdeprecated=True IncomingQuery：

from typing import List, Optional

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: list = Query(None, alias="name", deprecated=True)):
    query_items = {"q": q}
    return query_items

Path parameters and numerical verification

Path parameters

Path parameter{}The parameters of the parcel will be passed to your function as a parameter.

from fastapi import FastAPI

app = FastAPI()


@app.get("/test/{id}")
async def test(id):
    return {"id": id}

Type path parameters

from fastapi import FastAPI

app = FastAPI()


@app.get("/test/{id}")
 Async def test: # Defines the parameter to int type
    return {"id": id}

The above code will display the following information if access http: // localhost: 8000 / TEST / LOWELL.

{"detail":[{"loc":["path","id"],"msg":"value is not a valid integer","type":"type_error.integer"}]}

Only the input path ID can be accessed normally.

{"id":1}

Note that the value of the function receives (and returns) is 1, is a Pythonint Value, not a string"1"。

so,FastAPI The automatic «resolution» of the request is provided by the above type declaration. Automatic data conversion.

The order is very important

CreationPath operationWhen you find some cases it is fixed.

for example /users/meWe assume that it is used to obtain data about the current user.

Then you can use the path/users/{user_id} To obtain data about a particular user through the user ID.

due toPath operationIn order to run in order, you need to make sure the path/users/me Declare on the path/users/{user_id}Before:

from fastapi import FastAPI

app = FastAPI()


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

otherwise,/users/{user_id} The path will also be/users/me Match, «think» you are receiving a value"me" of user_id parameter.

default value

If you have a path operation of receiving path parameters, but you want to set a possible valid parameter value in advance, you can use standard Python.Enum Types of.

Create Enum kind

ImportEnum And create a inheritancestr and Enum Subclass.

Fromstr Inherited, the API document will be able to know that these values ​​must bestring Type and can be displayed correctly.

Then create class properties with fixed values, which will be available value:

from fastapi import FastAPI
from enum import Enum

app = FastAPI()


class Name(str, Enum):
    a = "hello"
    b = "lowell"
    c = "xxxx"


@app.get("/test/{name}")
async def test(name: Name):
    if name == Name.a:
        return {"name": name}
    elif name == Name.b:
        return {"name": name}
    return {"name": ""}

You can also passModelName.lenet.value Come get a value.

from fastapi import FastAPI
from enum import Enum

app = FastAPI()


class Name(str, Enum):
    a = "hello"
    b = "lowell"
    c = "xxxx"


@app.get("/test/{name}")
async def test(name: Name):
    print(Name.a.value) # hello
    if name == Name.a:
        return {"name": name}
    elif name == Name.b:
        return {"name": name}
    return {"name": ""}

Path converter

You can use the options directly from the Starlette to declare a containmentpathPath parameters:

/files/{file_path:path}

In this case, the name of the parameter isfile_pathAt the end:path Explain that this parameter should match anypath。

So you can use it like this:

from fastapi import FastAPI
from enum import Enum

app = FastAPI()


@app.get("/test/{name:path}")
async def test(name):
    return {"name": name}

You can access http: // localhost: 8000 / test / hello, normal display {«name»: «hello»}

You can also access Name as a path.Http: // localhost: 8000 / test // bin / downlaod / test, display {«name»: «/ bin / downlaod / test»}, the location of the path is double slash.

Declare metadata

You can declareQuery All parameters are the same.

For example, to declare path parametersitem_idof title Metadata value, you can enter:

from typing import List, Optional

from fastapi import FastAPI, Query
from fastapi.param_functions import Path

app = FastAPI()


@app.get("/items/{item_id}")
 Async Def Read_Items (item_id: int = path (..., title = "iTEM ID", Description = "ID", alias = "id"):
    return {"item_id": item_id}

Path parameters are always required because it must be part of the path.

So, you should use it when you declare... Mark it as a required parameter.

However, even if you useNone Declaration path parameters or set one other defaults will not have any effect, it will still be a required parameter.

Tips for parameters

If you want not to useQuery Declare query parameters without default valuesq,use simultaneously Path Declaration path parametersitem_idAnd make their order different from above, and Python has some special syntax.

Pass* As the first parameter of the function.

Python will not* Do anything, but it will know all the parameters of the following parameters should be used as a keyword parameter (key value), also known askwargsCall. Even if they don’t have a default.

from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    *, item_id: int = Path(..., title="The ID of the item to get"), q: str
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results

Numerical verification

greater or equal toge

use Query and Path(And other classes you will see later) can declare string constraints, but you can also declare numeric constraints.

Image below, addge=1 back,item_id Will be a big greater than (gREATER THAN) or equal toequal）1 Integer.

from typing import List, Optional

from fastapi import FastAPI, Query
from fastapi.param_functions import Path

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(item_id: int = Path(..., ge=1)):
    return {"item_id": item_id}

Less than or equalle

from typing import List, Optional

from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(item_id: int = Path(..., le=10)):
    return {"item_id": item_id}

more than thegtBe less thanlt

from typing import List, Optional

from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(item_id: int = Path(..., gt=2, lt=10)):
    return {"item_id": item_id}

Label

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/", response_model=Item, tags=["items"])
async def create_item(item: Item):
    return item


@app.get("/items/", tags=["items"])
async def read_items():
    return [{"name": "Foo", "price": 42}]


@app.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "johndoe"}]

Abstract

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/",
    response_model=Item,
    summary="Create an item",
    description="Create an item with all the information, name, description, price, tax and a set of unique tags",
)
async def create_item(item: Item):
    return item

Description of document strings

Description is often long, and override, you can functionDocumentationString declaration

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/", 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
    """
    return item

Response description

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/", response_model=Item, response_description="The created item")
async def create_item(item: Item):
    return item

No API interface

useinclude_in_schemaParameter, set to false

from fastapi import FastAPI

app = FastAPI()


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


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

This interface is not displayed when accessing the http: // localhost: 8000 / docs document interface.

API interface abandoned

If you need to mark the path operation as an abandoned, do not delete it, please pass the parameters:deprecated

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/", tags=["items"])
async def read_items():
    return [{"name": "Foo", "price": 42}]


@app.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "johndoe"}]


@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():
    return [{"item_id": "Foo"}]

ask

When you need to send data from a client (e.g., browser) to the API, you send it as a «request body».

askThe body is the data sent by the client to the API.responseThe body is the data sent to the client API.

Your API is almost always to sendresponsebody. But the client does not always need to sendaskbody.

We use Pydantic Model to declareaskBody and all the capabilities and advantages they have.

You can’t useGET Operation (HTTP method) sends a request body.

To send data, you must use one of the following methods:POST(More common),PUT、DELETE or PATCH。

First, you need topydantic IntroductionBaseModel：

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel


// json format transmission, equivalent to the format of JSON data
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):
    return item

As in declaring query parameters, it is not required when a model attribute has a default value. Otherwise it is a required property. Set the default value toNone It makes it an optional attribute.

Use model

In the function, you can directly access all the properties of the model object:

from typing import Optional
from fastapi import FastAPI
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()
    print(item_dict)
    print(item.name)
    print(item.description)
    return item

Single value in the request body

UseQuery and Path The way the query parameters and path parameters are defined in the same way.FastAPI Provide an equivalentBody。

For example, in order to expand the previous model, you may decideitem and user In addition, I want to have another button in the same request body.importance。

If you declare it as it is, because it is a single value,FastAPI Will assume that it is a query parameter.

But you can useBody instruct FastAPI Hand it as another key of the request body.

from typing import List, Optional

from fastapi import FastAPI, Path
from fastapi.param_functions import Body

app = FastAPI()


@app.post("/items/{item_id}")
async def read_items(item_id: int = Path(..., gt=2, lt=10), importance: int = Body(...)):
    return {"item_id": item_id, "importance": importance}

Embed a single request body parameter

Suppose you have only one from the Pydantic modelItem Request volume parametersitem。

by default,FastAPI Such a request body will be directly desired.

But if you want it to expect an owneditem The key and JSON containing the model content in the value, just like it is done when the additional requestor parameters are declared, you can use a specialBody Parameterembed：

item: Item = Body(..., embed=True)

for example:

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(..., embed=True)):
    results = {"item_id": item_id, "item": item}
    return results

under these circumstances,FastAPI A request body that will be expected to be like this:

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
}

Instead:

{
    "name": "Foo",
    "description": "The pretender",
    "price": 42.0,
    "tax": 3.2
}

Request body field

Declare model properties

Then you can use the model propertiesField：

from typing import Optional

from fastapi import Body, FastAPI
from pydantic import BaseModel
from pydantic.fields import Field

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = Field(None, title="The description of item", max_length=256)
         Price: float = field (none, ge = 0, description = "product price")
    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

Field Working method andQuery、Path and Body The same, including their parameters, etc. are also identical.

Request body — nested model

List field

You can define a property as a type with child elements. For example Pythonlist：

from typing import List, Optional

from fastapi import Body, FastAPI
from pydantic import BaseModel
from pydantic.fields import Field

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = Field(None, title="The description of item", max_length=256)
         Price: float = field (none, ge = 0, description = "product price")
    tax: Optional[float] = None
    tags: List = []


@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

List field with subtype

from typing import List, Optional

from fastapi import Body, FastAPI
from pydantic import BaseModel
from pydantic.fields import Field

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = Field(None, title="The description of item", max_length=256)
         Price: float = field (none, ge = 0, description = "product price")
    tax: Optional[float] = None
         # Defines a type of list field
    tags: List[str] = []


@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

Set type

But then we consider it, realize that the label should not be repeated, and they may be a unique string.

Python has a special data type to save a unique element, namelyset。

Then we can importSet Willtag Statement as a bystr consist of set：

from typing import List, Optional, Set

from fastapi import Body, FastAPI
from pydantic import BaseModel
from pydantic.fields import Field

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = Field(None, title="The description of item", max_length=256)
         Price: float = field (none, ge = 0, description = "product price")
    tax: Optional[float] = None
    tags: Set[str] = set()


@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

Nested model

from typing import List, Optional, Set

from fastapi import Body, FastAPI
from pydantic import BaseModel
from pydantic.fields import Field

app = FastAPI()


class Image(BaseModel):
    url: str
    name: str


class Item(BaseModel):
    name: str
    description: Optional[str] = Field(
        None, title="The description of item", max_length=256)
         Price: float = field (none, ge = 0, description = "product price")
    tax: Optional[float] = None
    tags: Set[str] = set()
    image: Optional[Image] = 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

this means FastAPI The request body similar to the following is desired:

{
  "item_id": 6,
  "item": {
    "name": "string",
    "description": "string",
    "price": 0,
    "tax": 0,
    "tags": [],
    "image": {
      "url": "https://www.baidu.com",
             "Name": "Baidu"
    }
  }
}

Extra information

you can use it Config and schema_extra Declare an example for the Pydantic model, such asPydantic Document: Customized SchemaAs described in:

from typing import List, Optional, Set

from fastapi import Body, FastAPI
from pydantic import BaseModel
from pydantic.fields import Field

app = FastAPI()


class Image(BaseModel):
    url: str
    name: str


class Item(BaseModel):
    name: str
    description: Optional[str] = Field(
        None, title="The description of item", max_length=256)
         Price: float = field (none, ge = 0, description = "product price")
    tax: Optional[float] = None
    tags: Set[str] = set()
    image: Optional[Image] = 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 = Body(..., embed=True)):
    results = {"item_id": item_id, "item": item}
    return results

These additional information will be added to the output JSON mode. It is equivalent to defining the default value.

Field Additional parameters

exist Field, Path, Query, Body Factory functions that will be seen after you will see you, you can declare additional information for JSON mode, you can also declare additional information to the JSON mode by delivering any arbitraries to factory functions, such as increaseexample:

from typing import List, Optional, Set

from fastapi import Body, FastAPI
from pydantic import BaseModel
from pydantic.fields import Field

app = FastAPI()


class Image(BaseModel):
    url: str
    name: str


class Item(BaseModel):
    name: str
    description: Optional[str] = Field(
        None, title="The description of item", max_length=256)
    price: float = Field(None, example=3.5)
    tax: Optional[float] = None
    tags: Set[str] = set()
    image: Optional[Image] = 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

Keep in mind that the extra parameters passing will not add any validation, only the comment is added, the purpose of the document.

Body Extra parameter

You can put one of the requested bodyexample Pass to Body:

from typing import List, Optional, Set

from fastapi import Body, FastAPI
from pydantic import BaseModel
from pydantic.fields import Field

app = FastAPI()


class Image(BaseModel):
    url: str
    name: str


class Item(BaseModel):
    name: str
    description: Optional[str] = Field(
        None, title="The description of item", max_length=256)
    price: float = Field(None, example=3.5)
    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

Use any method above, it/docs It looks like this:

Direct use request object

from fastapi import FastAPI, Request

app = FastAPI()


@app.get("/items/{item_id}")
def read_root(item_id: str, request: Request):
    client_host = request.client.host
    port = request.client.port
    url = request.url
    base_url = request.base_url
    headers = request.headers
    method = request.method
    path_params = request.path_params
    query_params = request.query_params
    state = request.state

    return {"item_id": item_id, "client_host": client_host, "port": port, "url": url, "base_url": base_url,
            "headers": headers, "method": method, "path_params": path_params, "query_params": query_params,
            "state": state}

Respond to body

{
  "item_id": "lowell",
  "client_host": "127.0.0.1",
  "port": 49432,
  "url": {
    "_url": "http://localhost:8000/items/lowell"
  },
  "base_url": {
    "_url": "http://localhost:8000/"
  },
  "headers": {
    "host": "localhost:8000",
    "connection": "keep-alive",
    "accept": "application/json",
    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.96 Safari/537.36 Edg/88.0.705.50",
    "sec-fetch-site": "same-origin",
    "sec-fetch-mode": "cors",
    "sec-fetch-dest": "empty",
    "referer": "http://localhost:8000/docs",
    "accept-encoding": "gzip, deflate, br",
    "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
    "cookie": "Pycharm-54243081=cd6e9076-310e-48d0-adb5-0c836dc36a30; fakesession=fake-cookie-session-value"
  },
  "method": "GET",
  "path_params": {
    "item_id": "lowell"
  },
  "query_params": {},
  "state": {
    "_state": {}
  }
}

Cookie

Get a cookie from the front end

To get a cookie, you must need to useCookieTo declare, otherwise the parameters will be interpreted as query parameters.

from typing import Optional

from fastapi import Cookie, FastAPI

app = FastAPI()


@app.get("/items/")
async def read_items(ads_id: Optional[str] = Cookie(None)):
    return {"ads_id": ads_id}

Set cookie

You can declare in the path operation functionType parameter。Response

Then you can respond to the object in this nativemiddleSet the cookie.

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

You can also create a cookie when you return to code.

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()


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

Get header information needs to use Header

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}

Access http: // localhost: 8000 / items / UA information will be displayed through the browser.

We will find that in the codeuser_agentAlong the request headerUser-AgentThe value, but their case and symbols are different. Why is this?

Most of the key in the header is used-To split, for exampleUser-AgentHowever, this name is not conforming to the norm in Python, soHeaderWill automatically underscore the parameter name_Convert to hyphen-. In addition, the HTTP request header is not case sensitive, so we can use a naming method compliant with the Python specification to represent them.

If you need to disable underscores for some reason_To even characters-Automatic conversion, you need to set your header’s parameters CONVERT_UNDERSCORES to FALSE:

from typing import Optional

from fastapi import FastAPI, Header

app = FastAPI()


@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None, convert_underscores=False)):
    return {"User-Agent": user_agent}

At this time we are requestinghttp://127.0.0.1:8000/items/When it is returned:

{"User-agent": null} # Because the underscore is not automatically converted -, it is not possible to get UA

Repeat the header information in the request

Repeated headers can be received. This means that there are multiple values ​​of the same header.

You can define these situations using the lists in the type declaration.

You will receive all values ​​as Python from the repeating header.list

For example, if you want to declare multiple headers, you can write:X-Token

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}

If you operate with this pathCommunication, sendTwo http headers, such as:

X-Token: foo
X-Token: bar

Response will be like:

{
    "X-Token values": [
        "bar",
        "foo"
    ]
}

Set head information when responding

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

You can also set the response head when you return directly.

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()


@app.get("/headers/")
def get_headers():
    content = {"message": "Hello World"}
    headers = {"X-Cat-Dog": "alone in the world", "Content-Language": "en-US"}
    return JSONResponse(content=content, headers=headers)

response

Returns the same data as the request

You can be in anyPath operationIn useresponse_model Parameters to declare the model used for response:

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Optional[str] = None


# Don't do this in production!
@app.post("/user/", response_model=UserIn)
async def create_user(user: UserIn):
    return user

Specify a response field

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()


class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Optional[str] = None


class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None


@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user

therefore,FastAPI Will be responsible for filtering out all the data declared in the output model

Exclude fields in response to fields

response_model_exclude_unset=True: Exclude the default field in response to returns only the field incoming actual value

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

Take the above response model as an example, if they do not store the actual value, you may want to use the field that displays the default value in response to useresponse_model_exclude_unset=TrueExclude unpreplified value fields. code show as below:

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_name}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_name: str):
    return items[item_name]

Accessing http: // localhost: 8000 / items / foo response will not contain those defaults, but only values ​​actually set. The results shown are as follows:

{
  "name": "Foo",
  "price": 50.2
}

Instead:

{
  "name": "Foo",
  "description": "string",
  "price": 50.2,
  "tax": 10.5,
  "tags": []
}

You can also use:

response_model_exclude_defaults=True: Whether to exclude the field equal to its default value from the returned dictionary, default is false

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, "tax": 10.5},
    "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_name}", response_model=Item, response_model_exclude_defaults=True)
async def read_item(item_name: str):
    return items[item_name]

Access http: // localhost: 8000 / items / foo will be displayed:

{
  "name": "Foo",
  "price": 50.2
}

Instead:

{
  "name": "Foo",
  "description": "string",
  "price": 50.2,
  "tax": 10.5,
  "tags": []
}
 // or 
{
  "name": "Foo",
  "price": 50.2,
  "tax": 10.5,
}

response_model_exclude_none=True: Whether to exclude from the returned dictionary equal to NONE field

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[str] = "a"
    tags: List[str] = []


items = {
    "foo": {"name": "Foo", "price": 18.0, "tax": None},
}


@app.get("/items/{item_name}", response_model=Item, response_model_exclude_none=True)
async def read_item(item_name: str):
    return items[item_name]

Access http: // localhost: 8000 / items / foo, will display

{
  "name": "Foo",
  "price": 18,
  "tags": []
}

Instead of

{
  "name": "Foo",
  "description": None,
  "price": 18,
  "tax": None,
  "tags": []
}

response_model_include : Display the specified field

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[str] = "a"
    tags: List[str] = []


items = {
    "foo": {"name": "Foo", "price": 18.0, "tax": "abc", "tags": ["lowell", "xiaoqi"]},
}


@app.get("/items/{item_name}", response_model=Item, response_model_include={"name", "price"})
async def read_item(item_name: str):
    return items[item_name]

response_model_exclude: Exclude the specified field

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[str] = "a"
    tags: List[str] = []


items = {
    "foo": {"name": "Foo", "price": 18.0, "tax": "abc", "tags": ["lowell", "xiaoqi"]},
}


@app.get("/items/{item_name}", response_model=Item, response_model_exclude={"name", "price"})
async def read_item(item_name: str):
    return items[item_name]

Suggest:It is recommended that you use the idea mentioned above, use multiple classes instead of these parameters. This is becauseresponse_model_include or response_model_exclude To omit certain properties, the JSON Schema generated in the OpenAPI definition (and documentation) will still be a complete model.

If you forget to useset But uselist or tupleFastApi will still convert it toset And normal work:

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[str] = "a"
    tags: List[str] = []


items = {
    "foo": {"name": "Foo", "price": 18.0, "tax": "abc", "tags": ["lowell", "xiaoqi"]},
}


@app.get("/items/{item_name}", response_model=Item, response_model_include=["name", "price"])
async def read_item(item_name: str):
    return items[item_name]

Union or anyOf

You can declare a response as two typesUnionThis means that the response will be any of two types. This will be used in OpenAPIanyOf Perform definition.

Define oneUnion When the type, first include the most detailed type, then it is less detailed. In the example below, more detailedPlaneItem lie in Union[PlaneItem，CarItem] middle CarItem Before.

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class BaseItem(BaseModel):
    description: str
    type: str


class CarItem(BaseItem):
    type = "car"


class PlaneItem(BaseItem):
    type = "plane"
    size: int


items = {
    "item1": {"description": "All my friends drive a low rider", "type": "car"},
    "item2": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "plane",
        "size": 5,
    },
}


@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem])
async def read_item(item_id: str):
    return items[item_id]

Define the response structure as a list

You can use the same way to declare the response made by the object list.

To do this, use standard Pythontyping.List：

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

Arbitrarydict Response

You can also use an arbitrary ordinarydict Declare response, only the type of key and value is declared without using the Pydantic model.

This will be useful if you don’t know the valid field / property name in advance (required for the Pydantic model).

In this case, you can usetyping.Dict：

from typing import Dict

from fastapi import FastAPI

app = FastAPI()


@app.get("/keyword-weights/", response_model=Dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

Return to a JSON directly

When you create oneFastAPI Path operation You can return any of the following data normally:dict，list, Pydantic model, database model, etc.

FastAPI By defaultjsonable_encoder Convert these types of return values ​​to JSON format,jsonable_encoder exist JSON compatible encoder There is an explanation.

Then,FastAPI Put these compatible JSON data (such as dictionary) in the backgroundJSONResponse Medium, thisJSONResponse Will use to send a response to the client.

But you can be in yourPath operation Return one directlyJSONResponse。

When you use it directlyJsonResponsewhen,FastAPI No data conversion is made with a PyDantic model, and the response content will not be converted into any type. Need us to use manualjsonable_encoderConvert data model

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)

Return to custom Response

Suppose you want to return oneXML response.

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

When you return directlyResponse When its data has neither verified, it will not be converted (serialized), nor does it automatically generate documents.

Documents and overload in OpenAPIResponse

Returned directlyHTMLResponse

For example, like this:

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 you areresponse_class Also introducedHTMLResponse，FastAPI Will know how to use in OpenAPI and interactive documentstext/html Will document its document to HTML.

Available response

Response

All other responses have inherited their own classResponse。

You can return it directly.

Response The class accepts the following parameters:

• content — One str or bytes。
• status_code — One int Type HTTP status code.
• headers — A string consisting of stringsdict。
• media_type — A given media typestr,for example "text/html"。

FastApi (actually starlette) will automatically contain the header of Content-Length. It will also contain a media_type’s Content-Type header and add a character set for text types.

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

HTMLResponse

Accept text or byte and return to HTML response

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()


@app.get("/items/")
async def read_items():
    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)

PlainTextResponse

Accept text or bytes and return to a plain text response.

from fastapi import FastAPI
from fastapi.responses import PlainTextResponse

app = FastAPI()


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

JSONResponse

Accept data and return oneapplication/json The response of the encoding.

ORJSONResponse

If you need a press performance, you can install and use it.orjson And set the response toORJSONResponse。

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

UJSONResponse

UJSONResponse A useujson Optional JSON response.

from fastapi import FastAPI
from fastapi.responses import UJSONResponse

app = FastAPI()


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

RedirectResponse

Returns the HTTP redirection. 307 status code (temporary redirection) is used by default.

from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()


@app.get("/typer")
async def read_typer():
    return RedirectResponse("https://typer.tiangolo.com")

StreamingResponse

The asynchronous generator or ordinary generator / iterator is used, and then the streaming response body is used.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"


@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer())

If you have objects like files (for example, byopen() The object returned), thenStreamingResponse It returns it in the middle.

Includes many libraries interact with cloud storage, video processing, etc.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
def main():
    file_like = open(some_file_path, mode="rb")
    return StreamingResponse(file_like, media_type="video/mp4")

FileResponse

Asynchronous transmission files act as a response.

Accept different parameter sets for instantiation compared to other response types:

• path — The file path of the file to be transferred.
• headers — Any custom response header, incoming dictionary type.
• media_type — Give a string of the media type. If not set, the file name or path will be used to infer the media type.
• filename — If given, it will be included in responseContent-Disposition middle.

The file response will contain the appropriateContent-Length，Last-Modified and ETag Response head.

from fastapi import FastAPI
from fastapi.responses import FileResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


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

Additional responses in OpenAPI

Ordinary JSON response

FastAPIThis model will be accepted to generate its JSON architecture, and it contains it in the correct location in OpenAPI.

For example, if you want to use a status code and a Pydant model to declare another response, you can write:404``Message

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

Remember, you have to return directly.JSONResponse

404 errors will be displayed. Or access /openapi.json can also access the 404 error message.

Media type response

Other media types, statement paths can be addedOperationReturns the JSON object (with a media type) or a png image:image/png application/json

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("image.png", media_type="image/png")
    else:
        return {"id": "foo", "value": "there goes my hero"}

Notice:Media type must be usedFileResponse

Response status code

Can be used in path operationsstatus_code Parameters to declare the HTTP status code for the response:

from fastapi import FastAPI, status

app = FastAPI()


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

Can be usedfastapi.status Convenient variable.

from fastapi import FastAPI, status

app = FastAPI()


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

They are just a convenient way, they have the same digital code, but this can be used to use the editor’s automatic full function to find them.

Additional status code

If you want to return a status code outside of the main status code, you can return one directlyResponse To achieve, for exampleJSONResponse, Then set the additional status code directly.

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse

app = FastAPI()

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


@app.get("/items/{item_id}")
async def upsert_item(item_id: str):
    if item_id in items:
        return {"item_id": item_id}
    else:
        return JSONResponse(status_code=status.HTTP_201_CREATED, content={"item_id": item_id})


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

At this point, the access address will return 201’s status code.

Modify the status code when the response

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]

Form data

If the front end is not JSON format data, but the Form form data. Get data in the Form form via form ()

Use the Form form to first installpython-multipart

pip install python-multipart

Code example:

from fastapi import FastAPI, Form

app = FastAPI()


@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

The front end Enter username and password via the Form form will receive form data via Form.

Transfer file

Transfer files need to download and installpython-multipart

pip install python-multipart

upload files

from fastapi import FastAPI, File, UploadFile
from starlette.routing import Host
from uvicorn import config

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 {"filename": file.filename}

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

UploadFileHas the following properties:

• filename: With the original file name uploaded (for example).str``myimage.jpg
• content_type: A (for example) having content type (MIME type / media type).str``image/jpeg
• file：Backstage document (documentClass object). This is the actual Python file, you can pass directly to other functions or libraries that need the «File» object.

UploadFileThere are the following methods. They all call the following corresponding file method (use interior).async«SpooledTemporaryFile

• write(data): Write (or) to the file.data``str``bytes
• read(size): Read () the byte / character of the file.size``int
• seek(offset): Go to the byte position in the file (). Offset int
  • For example, it will go to the beginning of the file.await myfile.seek(0)
  • This is especially useful if you are running again and then need to read content again.await myfile.read()
• close(): Close the file.

For example, in the path operationFunctionYou can get the following:async

contents = await myfile.read()

If you are in normal path operationFunction within the functionYou can access directly, for example:def``UploadFile.file

contents = myfile.file.read()

Multiple file upload

To use it, please declare or:List bytes UploadFile

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

Error handling

In many cases, you need to notify the error using the API client. FastApi returns an error message via HTTPEXCEPTION

Use HTTPEXCEPTION

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

Add a custom error response

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

Customization process

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}

JSON Compatibility Coding

When you need to convert JSON data to data types such as List / DICT, we need to use JSON Compatibility Code Modules JSonable_Encoder module

from os import name
from typing import Optional
from fastapi import FastAPI
from datetime import datetime

from fastapi.encoders import jsonable_encoder
from pydantic.main import BaseModel


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


app = FastAPI()


@app.put("/items/{id}")
def update_item(id: int, item: Item):
    print(type(item)) # <class 'main.Item'>
    json_compatible_item_data = jsonable_encoder(item)
    print(type(json_compatible_item_data)) # <class 'dict'>
    return json_compatible_item_data

Dependency injection

Function dependent

The parameters use the parameters of the dependencies that are dependent on the function for multiple functions.

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

Class as a dependency

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

Additional writing

commons: CommonQueryParams = Depends()

Sub-dependence

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}

Path operation modifier

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

The execution / solution of these dependencies is the same as the normal dependency. But their values ​​(if returned any value) will not pass to your pathOperation function,Right nowread_items。

This dependency can be understood as an essential condition to access the path, and the front end does not carry two parameters cannot access this path.

Global dependence

Add dependence to the entire application, all paths depend

from fastapi import Depends, FastAPI, Header, HTTPException


async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])


@app.get("/items/")
async def read_items():
    return [{"item": "Portal Gun"}, {"item": "Plumbus"}]


@app.get("/users/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

Dependencies with yield

For details, please see the official website https://fastapi.tiangolo.com/zh/tutorial/dependencies/dependencies-with-yield/#using-context-Managers-in-dependencies-with-yield

You must use Yield, Python version3.7+If you use python3.6 to install the dependency library, you are as follows

pip install async-exit-stack async-generator

This is an example of a dependency function. We can create a database session and close this session after the request is completed.

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

yield dbBack valuedbThe path operation or other dependencies will be injected.

yield dbBehind the coderesponseIt will be performed after submission.

Middleware

«Middleware» is a function, it can pass any specificPath operationHandle eachaskPreviously handled eachask. And return before returningresponse。

If you have a dependencyyieldExit the codeexistMiddlewareLaterrun.

If there is any background task, they willexistAll middlewareLaterrun.

Create a middleware

To create a middleware, please@app.middleware("http")Use a decorator at the top of the function.

Intermediate piece function parameters:

• request
• CALL_NEXT function: Will receive a request as a parameter, return a response object response of a current path, can modify the Response, and then return.

import time

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

You can also usefrom starlette.requests import Request。

FastAPIProvided by developers. But it comes directly from Starlette.

FastAPI built-in middleware

HTTPSRedirectMiddleware

Forced all incoming requests must be https or WSS, any incoming HTTP or WS request will redirect to https or WSS

from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

app.add_middleware(HTTPSRedirectMiddleware)


@app.get("/")
async def main():
    return {"message": "Hello World"}

TrustedHostMiddleware

Forced all incoming requests have a correct set of heads to prevent HTTP host attacks

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()

app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"]
)


@app.get("/")
async def main():
    return {"message": "Hello World"}

GZipMiddleware

Process the Gzip response included in the request header information,

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

app.add_middleware(GZipMiddleware, minimum_size=1000)


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

minimum_size: Do not perform GZIP processing on the response smaller than this minimum byte size. Default 500

CORS cross-domain resource sharing

You can use itFastAPIConfigure it in the applicationCORSMiddleware。

• ImportCORSMiddleware。
• Create a list of allowable sources (as a string).
• Add it as a «middleware» toFastAPIapplication.

You can also specify whether the backend is allowed:

• Voucher (Authorized Header, Cookie, etc.).
• Specific HTTP method (POST，PUT) Or all bands"*"。
• Specific HTTP header or all wildcard"*"。

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

ShouldCORSMiddlewareThe default parameters for implementations are restrictive by default, so you need to explicitly enable specific sources, methods, or headers to allow the browser to use them in cross-domain contexts.

Support the following parameters:

• allow_origins— The source list of cross-domain requests should be permitted. E.g['https://example.org', 'https://www.example.org']. you can['*']Used to allow any sources.

• allow_origin_regex— A regular expression string that matches the origin that should be allowed to perform cross-domain requests. E.g'https://.*.example.org'。

• allow_methods— The HTTP method for cross-domain requests should be allowed. The default is['GET']. you can use it['*']Allow all standard methods.

• allow_headers— HTTP request header list should be supported by cross-domain requests. The default is[]. you can['*']Used to allow all headers.Accept，Accept-Language，Content-LanguageandContent-TypeHead always allows CORS requests.

• allow_credentials— Indicates that cross-domain requests should support cookies. The default isFalse. in addition,allow_originsCannot set up['*']For allowable credentials, the source must be specified.

• expose_headers— It is pointed out that the browser should be accessed. The default is[]。

• max_age— Set the maximum time (in seconds) of the browser cache CORS response. The default is600

You can also usefrom starlette.middleware.cors import CORSMiddleware。

FastAPIfastapi.middlewareSeveral middleware is provided for your (developer). But most of the available middleware is directly from Starlette.

Project development writing

If you want to build an application or web API, you will rarely place all content in a single file.FastAPIProvide a convenient tool to build an application while maintaining all flexibility.

Similar to the blueprint of Flask.

File structure directory

Suppose you have a file structure:

.
├── app
│   ├── __init__.py
│   ├── main.py
│   ├── dependencies.py
│   └── routers
│   │   ├── __init__.py
│   │   ├── items.py
│   │   └── users.py
│   └── internal
│       ├── __init__.py
│       └── admin.py

Similar to Flask’s blueprint, unputtoping modules to facilitate understanding.

Sample program

main.py

from fastapi import Depends, FastAPI

from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])

app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)


@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

dependencies.py

from fastapi import Header, HTTPException


async def get_token_header(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No Jessica token provided")

routers.items.py

from fastapi import APIRouter, Depends, HTTPException

from ..dependencies import get_token_header

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)

fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}


@router.get("/")
async def read_items():
    return fake_items_db


@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"name": fake_items_db[item_id]["name"], "item_id": item_id}


@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
    if item_id != "plumbus":
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus"
        )
    return {"item_id": item_id, "name": "The great Plumbus"}

routers.users.py

from fastapi import APIRouter

router = APIRouter()


@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]


@router.get("/users/me", tags=["users"])
async def read_user_me():
    return {"username": "fakecurrentuser"}


@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}

internal.admin.py

from fastapi import APIRouter

router = APIRouter()


@router.post("/")
async def update_admin():
    return {"message": "Admin getting schwifty"}

starting program:

uvicorn app.main:app --reload

Background task

The background task is useful for the operations that need to be executed after the request, but the client does not have to wait for the operation before receiving the response.

This includes, for example:

• Email notification sent after performing operation:
  • Since the e-mail server and send email are often «slow» (a few seconds), you can return to respond immediately and send email notifications in the background.
• Processing data:
  • For example, assuming that the file you receive must pass through a slow process, you can return the response of «HTTP 202) and process it in the background.

Create a task function

Create a function to run as a background task.

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

.add_task()Received parameters

• write_notification: Task function to run in the background

• email: Parameters to the task function should be delivered in order

• message="some notification": Parameters to the task function should be delivered in order

If you need to perform a heavy background calculation, it is not necessarily to run by the same process (for example, you don’t need sharing memory, variables, etc.), use other more tools (such as Celery)

Metadata and document URL

Create metadata for project

• title: Used as the title / name of the API in the OpenAPI and Auto API Documentation UI.
• describe: Description of the API in the OpenAPI and Automatic API Document UI.
• Version: API version, for example: v2 2.5.0
  • For example, if you have an earlier version of an application, you also use OpenAPI, which is very useful.

To set them, use the parameters and:title description version

from fastapi import FastAPI

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",
)


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

Create metadata for tag

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(openapi_tags=tags_metadata)


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

OpenAPI URL

By default, the OpenAPI architecture is available in it./openapi.json

However, you can configure it using parameters.openapi_url

For example, set it to:/api/v1/openapi.json

from fastapi import FastAPI

app = FastAPI(openapi_url="/api/v1/openapi.json")


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

If you want to completely disable the OpenAPI architecture, you can set it, which will also disable the document user interface using it.openapi_url=None

Document URLS

passdocs_urlandredoc_urlSet the document URL, you can also set to NONE forbidden access

from fastapi import FastAPI

app = FastAPI(docs_url="/documentation", redoc_url=None)


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

Static file

Use static files to downloadaiofilesLibrary

pip install aiofiles

Use template files to be installedjinja2

pip install jinja2

Application example:

from fastapi import FastAPI, Form
from starlette.requests import Request
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory="templates")
app.mount('/static', StaticFiles(directory='static'), name='static')


@app.post("/user/")
async def files(
        request: Request,
        username: str = Form(...),
        password: str = Form(...),
):
    print('username', username)
    print('password', password)
    return templates.TemplateResponse(
        'index.html',
        {
            'request': request,
            'username': username,
        }
    )


@app.get("/")
async def main(request: Request):
    return templates.TemplateResponse('signin.html', {'request': request})

test

Ordinary test

Test relies on Requests, so you need to install

pip install requests

Use tests need to be installedpytest

pip install pytest

Suppose you have an includedFASTAPI applicationdocument:main.py

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

Test file

from fastapi.testclient import TestClient

from .main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

Execute test files

pytest

Asynchronous test

Install asynchronous test library

pip install pytest-asyncio

Need HTTPX library

pip install httpx

For a simple example, let us consider the following modules:main.py

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Tomato"}

The module containing is now like this:test_main.py``main.py

import pytest
from httpx import AsyncClient

from .main import app


@pytest.mark.asyncio
async def test_root():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

Sub-application

First, create a master, top、FastAPIApplication and itsPath operation：

from fastapi import FastAPI

app = FastAPI()


 #Tlet-level application
@app.get("/app")
def read_main():
    return {"message": "Hello World from main app"}


subapi = FastAPI()

 # 子 应用
@subapi.get("/sub")
def read_sub():
    return {"message": "Hello World from sub API"}

 # Loading
app.mount("/subapi", subapi)

Main application Access http: // localhost: 8000 / DOCS

Sub-apps Access http: // localhost: 8000 / Subapi / DOCS

websocket

Example

from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""


@app.get("/")
async def get():
    return HTMLResponse(html)


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

Process disconnect and multiple clients

When the WebSocket connection is turned off, an exception will be triggered, and then you can capture and process the exception, just as shown in this example.await websocket.receive_text()``WebSocketDisconnect

from typing import List

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <h2>Your ID: <span id="ws-id"></span></h2>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var client_id = Date.now()
            document.querySelector("#ws-id").textContent = client_id;
            var ws = new WebSocket(`ws://localhost:8000/ws/${client_id}`);
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""


class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)


manager = ConnectionManager()


@app.get("/")
async def get():
    return HTMLResponse(html)


@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_personal_message(f"You wrote: {data}", websocket)
            await manager.broadcast(f"Client #{client_id} says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client #{client_id} left the chat")

Event: Start — close

Event handlers (functions) that can be executed before or before the application is started or before the application is turned off

startupevent

To add functions that should be run before the application starts, use the event to declare it:"startup". Multiple event handler functions can be added.

from fastapi import FastAPI

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]

shutdownevent

To add a function that should be run when the application is turned off, use the event to declare it:"shutdown"

from fastapi import FastAPI

app = FastAPI()


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


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

Profile and environment variable

Environment variable

Read system environment variable

import os

name = os.getenv("MY_NAME", "World")
print(f"Hello {name} from Python")

Profile

The first:This way creates a global configuration object

Create config.py

from pydantic import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str
    items_per_user: int = 50


settings = Settings()

Then use it in the file:main.py

from fastapi import FastAPI

from . import config

app = FastAPI()


@app.get("/info")
async def info():
    return {
        "app_name": config.settings.app_name,
        "admin_email": config.settings.admin_email,
        "items_per_user": config.settings.items_per_user,
    }

Second:Do not create a default instance

From the previous example, your file may look like:config.py

from pydantic import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str
    items_per_user: int = 50

Please note that we don’t create a default instance now.settings = Settings()

Now we create a new dependency.config.Settings()

Then we can ask it fromPath operation functionAs a dependency, and use it anywhere, we need it.

from functools import lru_cache

from fastapi import Depends, FastAPI

from . import config

app = FastAPI()


@lru_cache()
def get_settings():
    return config.Settings()


@app.get("/info")
async def info(settings: config.Settings = Depends(get_settings)):
    return {
        "app_name": settings.app_name,
        "admin_email": settings.admin_email,
        "items_per_user": settings.items_per_user,
    }

@lru_cache(): When we use the @LRU_CACHE () modifier, the object will only create one time, otherwise we will create an object each time you call GET_SETTINGS.

Read .env file

You can have a file whose content is:.env

ADMIN_EMAIL="[email protected]"
APP_NAME="ChimichangApp"

Update to config.py file

from pydantic import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str
    items_per_user: int = 50

    class Config:
        env_file = ".env"