FastAPI勉強メモ

参考資料

FastAPIは？

FastAPI framework, high performance, easy to learn, fast to code, ready for production

• 特徴
  • typeの定義でrequest内容を自動bindingできる
  • type hintで静的チェックできるので、バグになりにくい
  • 簡単
  • ハイパフォーマンス
  • OAuth2
  • OpenAPI docs自動作成
  • GraphQLサポート
  • ドキュメントが充実

開発環境の構築

libraryインストール

• virtualenv is a tool for creating isolated virtual python environments.
  • プロジェクトごとで依存するパッケージの管理環境を提供する
• FastAPI is a framework, high performance, easy to learn, fast to code, ready for production
• Uvicorn is a lightning-fast ASGI server.

pip3 install virtualenv --user
mkdir fastapi-demo # project directoryを作成
python3 -m venv env # project 仮想環境を作成
source env/bin/activate # 仮想環境をアクティブする
pip3 install fastapi
pip3 install uvicorn

vscode extensionsインストール

• Python
• pylance : 静的型チェック、 コード入力補助機能を提供する
  • https://qiita.com/Tatejimaru137/items/94d84848d74c99b59157

python typeを理解する

Hello Worldを作ってみる

• main.pyファイルを作成して、以下のコードを書いて、保存する

# FastAPI is a Python class that provides all the functionality for your API.
from fastapi import FastAPI # 

app = FastAPI()

@app.get("/") # routing
# It will be called whenever it receives a request to the PATH "/" using a GET operation.
async def root():
    return {"message": "Hello World"}

• serverを起動

uvicorn main:app --reload # mainはファイル名前、 appはapp = FastAPI()での変数名と同じ

自動で作成されたAPI docs

• Interactive(swagger-ui) OpenAPI docs： http://127.0.0.1:8000/docs
• redocs: http://127.0.0.1:8000/redoc
• raw OpenAPI schema(openapi.json): http://127.0.0.1:8000/openapi.json

routingの定義（path operation decorator)

path + operation(http methods)でroutingを定義する

• サポートするoperation
  • get
  • post
  • put
  • patch
  • delete
  • options
  • head
  • trace

Path parameters

@app.get("/items/{item_id}")
async def read_item(item_id: int): # 変数item_idの名前は/items/{item_id}の{item_id}と一致する必要がある
    return {"item_id": item_id}

Path parametersのtypeを指定するメリット

• タイプの変換: path parameterの値から指定したtypeに自動で変換してくれる
• validation: path parameterの値が指定したtypeに合わない場合、validationエラーresponseを返す
• API docが分かり易くなる

Enum path parameter type

from enum import Enum

from fastapi import FastAPI

class Gender(str, Enum):
    male = "male"
    female = "female"

app = FastAPI()

@app.get("/models/{gender}")
async def get_model(gender: Gender):
    if gender == Gender.male:
        return {"gender": gender, "message": "Man!"}

    if gender.value == "female":
        return {"gender": gender, "message": "Woman!"}

Path convertor

from fastapi import FastAPI

app = FastAPI()

## http://127.0.0.1:8000/files//home/tom/main.pyをアクセスすると、
## file_pathの値が/home/tom/main.pyになります
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

Pathタイプを明示的にする

@app.get("/items/{item_id}")
async def read_items(
    item_id: int = Path(..., title="The ID of the item to get", gt=0, le=1000)
):

Query Parameters

# http://127.0.0.1:8000/items/?skip=0&limit=3
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

Optional parameters

from typing import Optional

@app.get("/items/{item_id}")
async def read_item(item_id: str, id: Optional[int] = None):

bool型のQuery parameter type

# http://127.0.0.1:8000/items/foo?short=1
# http://127.0.0.1:8000/items/foo?short=True
# http://127.0.0.1:8000/items/foo?short=true
# http://127.0.0.1:8000/items/foo?short=on
# http://127.0.0.1:8000/items/foo?short=yes
@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Optional[str] = None, short: bool = False):

Required query parameters

# http://127.0.0.1:8000/items/foo-itemをアクセスすると、
# needyの値がないので、validation errorになります
@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):

Query Parametersとstring Validations

from fastapi import FastAPI, Query

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

required query parameter(...)

@app.get("/items/")
async def read_items(q: str = Query(..., min_length=3)):

複数値のQuery parameter

# http://localhost:8000/items/?q=foo&q=bar
# you need to explicitly use Query, otherwise it would be interpreted as a request body.
@app.get("/items/")
async def read_items(q: Optional[List[str]] = Query(None)):

@app.get("/items/")
async def read_items(q: List[str] = Query(["foo", "bar"])):

Query parameter metadata

@app.get("/items/")
async def read_items(
    q: Optional[str] = Query(
        None,
        title="Query string",
        description="Query string for the items to search in the database that have a good match",
        min_length=3,
    )
):

Alias parameters

# http://127.0.0.1:8000/items/?item-query=foobaritems
@app.get("/items/")
async def read_items(item_query: Optional[str] = Query(None, alias="item-query")):

Deprecating parameters（非推奨parameter）

@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, deprecated=True)):

JSON Body parameters

Pydantic modelのparameterはrequestのJSON Bodyからデータバインディングする

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

で以下のようなjson bodyを受け取ることができる

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

parameterの名前をjson keyにしたいことにembed=Trueを付ける

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item = Body(..., embed=True)):

で以下のようなjson bodyを受け取ることができる

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

Multiple body parameters

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

class User(BaseModel):
    username: str
    full_name: Optional[str] = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
    results = {"item_id": item_id, "item": item, "user": user}
    return results

で以下のようなjson bodyを受け取ることができる

// key名前が、`async def update_item`のparameterと一致する必要がある
{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    }
}

Singular values in body

@app.put("/items/{item_id}")
async def update_item(
    item_id: int, importance: int = Body(...)
):

で以下のようなjson bodyを受け取ることができる

{
    "importance": 5
}

json body fieldのvalidation & metadata

from pydantic import Field

class Item(BaseModel):
    price: float = Field(..., gt=0, description="The price must be greater than zero", example="Foo")

json Body - Nested Models

from typing import Set

class Item(BaseModel):
    name: str
    tags: Set[str] = set()

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

class Item(BaseModel):
    name: str
    image: Optional[Image] = None

array list

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

@app.post("/images/multiple/")
async def create_multiple_images(images: List[Image]):

任意キーのjson body

from typing import Dict
from fastapi import FastAPI

app = FastAPI()

@app.post("/index-weights/")
async def create_index_weights(weights: Dict[int, float]):
    return weights

pydantic types

Form Data

formデータを処理するために、python-multipartをinstallする必要がある

pip3 install python-multipart

from fastapi import FastAPI, Form

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

Cookie parameters

from fastapi import Cookie, FastAPI

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

Header parameters

from fastapi import FastAPI, Header

@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):

重複するheader取得

@app.get("/items/")
async def read_items(x_token: Optional[List[str]] = Header(None)):

Response Model(serialization)

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserIn(BaseModel):
    password: str
    email: EmailStr

class UserOut(BaseModel):
    email: EmailStr

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

responseで複数のModel(serialization)を使いたい時、Unionで複数のmodelを指定する

from typing import Union

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

任意のjsonを返す時、Dictを使う

@app.get("/keyword-weights/", response_model=Dict[str, float])

共通の項目をもつbaseクラスを作ることで重複を減らす

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

class UserIn(UserBase):
    password: str

class UserOut(UserBase):
    pass

class UserInDB(UserBase):
    hashed_password: st

細かい設定

• response_model_include
• response_model_exclude
• response_model_exclude_unset
• response_model_exclude_defaults
• response_model_exclude_none
• response_model_by_alias

response status code指定

from fastapi import FastAPI, status

@app.post("/items/", status_code=status.HTTP_201_CREATED)
def get_or_create_task(task_id: str, response: Response):
    response.status_code = status.HTTP_204_NO_CONTENT # 明示的に指定する
    return tasks[task_id]

エラー処理

HTTPExceptionをraiseしてエラーをclientに返す

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",
            headers={"X-Error": "There goes my error"},
        )
    return {"item": items[item_id]}

global Exceptionハンドラー

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)}")
    # 元のhandlerを呼ぶ
    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}")
    # 元のhandlerを呼ぶ
    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}

FastAPI's HTTPException vs Starlette's HTTPException

• FastAPI's HTTPExceptionはFastAPI's HTTPExceptionを継承している
• raiseするときは、FastAPI's HTTPExceptionを使う
• handleするときは、Starlette's HTTPExceptionを使う

Path Operation Configuration(docs metadata)

@app.post(
    "/items/",
    response_model=Item,
    status_code=status.HTTP_201_CREATED,
    summary="Create an item",
    description="Create an item with all the information, name, description, price, tax and a set of unique tags",
    tags=["items"],
    response_description="The created item",
    deprecated=True,
)
async def create_item(item: Item):
    """
    This is docstring: 
    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
    """

Dependency Injection

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 dependencies

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(params: CommonQueryParams = Depends(CommonQueryParams)):
    response = {}
    if params.q:
        response.update({"q": params.q})
    items = fake_items_db[params.skip : params.skip + params.limit]
    response.update({"items": items})
    return response

shortcut

# below is shortcut of `async def read_items(params: CommonQueryParams = Depends(CommonQueryParams)):`
async def read_items(params: CommonQueryParams = Depends()):

実行しますが、戻り値を使わないdependencyはdependenciesで渡す

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

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

Parameterized dependencies

from fastapi import Depends, FastAPI

app = FastAPI()

class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False

@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(FixedContentQueryChecker("bar1"))):
    return {"fixed_content_in_query": fixed_content_included}

Global Dependencies

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

responseを返した後に、その他処理を行い時に、yieldを使う

async def get_db():
    db = DBSession()
    try:
        yield db
        # responseを返した後の処理
    finally:
        db.close()

或いはcontext managerを使う

class MySuperContextManager:
    def __init__(self):
        self.db = DBSession()

    def __enter__(self):
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()

async def get_db():
    with MySuperContextManager() as db:
        yield db

authorization

OAuth2PasswordBearer

• usernameとpasswordをform dataで送って、ログインして、Bearer tokenを発行する認証方法
• scopeを送ることでできる（https://fastapi.tiangolo.com/tutorial/security/simple-oauth2/#scope)

from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "[e-mail]",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "[e-mail]",
        "hashed_password": "fakehashedsecret2",
        "disabled": True,
    },
}

app = FastAPI()

def fake_hash_password(password: str):
    return "fakehashed" + password

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # usernameとpasswordをtoken urlに送って、tokenを貰う

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def fake_decode_token(token):
    # This doesn't provide any security at all
    # Check the next version
    user = get_user(fake_users_db, token)
    return user


async def get_current_user(token: str = Depends(oauth2_scheme)): # Authorization Headerからtokenを取得
    user = fake_decode_token(token)
    if not user:
        # The additional header WWW-Authenticate with value Bearer we are returning here is also part of the spec.
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    user = UserInDB(**user_dict)
    hashed_password = fake_hash_password(form_data.password)
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    return {"access_token": user.username, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

OAuth2PasswordBearer, Bearer with JWT tokens

JWT紹介：https://jwt.io/introduction/

pip install python-jose cryptography # generate and verify the JWT tokens
pip install passlib bcrypt # passwordのhash化で使う

from datetime import datetime, timedelta
from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "[e-mail]",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires # The JWT specification says that there's a key sub, with the subject of the token.
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

@app.get("/users/me/items/")
async def read_own_items(current_user: User = Depends(get_current_active_user)):
    return [{"item_id": "Foo", "owner": current_user.username}]

Middleware

requestを受け取って、事前処理をしたり、responseを返す前に後の処理を行ったりすることができる

@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

CORS(Cross-Origin Resource Sharing)

Origin

https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS
Cross-Originの判定はprotocol, domain, portの何が異なる場合

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

# parameters: https://fastapi.tiangolo.com/tutorial/cors/#use-corsmiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True, # https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Access-Control-Allow-Credentials
    allow_methods=["*"],
    allow_headers=["*"],
)

大きいプロジェクトの構成

APIRouter

app/routers/items.py

from fastapi import APIRouter, Depends, HTTPException

from ..dependencies import check_token_header

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(check_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

app/dependencies.py

from fastapi import Header, HTTPException

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

app/main.py

from fastapi import Depends, FastAPI

from .dependencies import check_query_token
from .routers import items

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

app.include_router(items.router)

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

Background Tasks

BackgroundTasksはrequestを処理する同じprocessを使うので、重いtasksの場合は、専用のqueueを使った方がいい。https://docs.celeryproject.org/en/stable/

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

Static Files(静的ファイル)

pip install aiofiles

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")

testing

pip install pytest
pip install requests

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

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

client = TestClient(app)
# clientのAPI：　https://requests.readthedocs.io/en/latest/api/

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

responseを直接返す

json responseを返す１

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

class Item(BaseModel):
    title: str
    description: str

app = FastAPI()

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

json responseを返す２

response_classで指定すると、OpenAPI docsが自動で生成される

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

html responseを返す

@app.get("/items/", response_class=HTMLResponse)
async def read_items():
    return """
    <html>
        <head>
            <title>Some HTML in here</title>
        </head>
        <body>
            <h1>Look ma! HTML!</h1>
        </body>
    </html>
    """

他のresponse class

• RedirectResponse
• FileResponse
• StreamingResponse

response_model, responsesでOpenAPI docsを生成

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

Debugging

以下のようなファイルを用意し、IDEでbreakpointを設定すれば、debugできます。

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    a = "a"
    b = "b" + a
    return {"hello world": b}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000) # コードでuviconを起動