はじめての FastAPI 後編

はじめての FastAPI（前編）
https://qiita.com/uturned0/items/9c3d19c1f846a07ec779

の続き

Handling Errors¶

HTTPException

http://0.0.0.0:8000/err/1 にいくと 404 が返る。raise するだけで json がレスポンスになるの最高。

from fastapi import FastAPI, Body, Query, HTTPException

items = {"foo": "The Foo Wrestlers"}
@app.get("/err/{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]}

custom error handling

特別なエラーに UnicornException て名前つけて、それが発生した時に必ず 418 返すようにする。

http://0.0.0.0:8000/unicorns/yolo　を開いたらちゃんと I'm a teapot が返ってきました

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}

override exception handlers

ここtutorialよくわからなかったんだけどこういうことだと思う

from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse([1])

errorを全部どっかに飛ばしたいとか、そういう処理を挟めるんだと思う

その後は debug 用のreturn を返す方法とか書いてんだけど、まあ、いったんskip

Re-use FastAPI's exception handlers¶

複数のエラー処理をしたいときはchain的にすることもできる。

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}

http://0.0.0.0:8000/items/foo だとvalidation errorが走って

API

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

console

OMG! The client sent invalid data!: 1 validation error for Request
path -> item_id
  value is not a valid integer (type=type_error.integer)
INFO:     127.0.0.1:61432 - "GET /items/foo HTTP/1.1" 422 Unprocessable Entity

http://0.0.0.0:8000/items/3 だと raise 418 が走って

api

{
detail: "Nope! I don't like 3."
}

console

OMG! An HTTP error!: HTTPException(status_code=418, detail="Nope! I don't like 3.")
INFO:     127.0.0.1:62125 - "GET /items/3 HTTP/1.1" 418 

ってことだと思う、多分

Database

普通の sqlalchemy の説明で、SQL使わないから飛ばそうかと思ったけど、ここ気になった

SQLAlchemy style and Pydantic style¶

Notice that SQLAlchemy models define attributes using =, and pass the type as a parameter to Column, like in:
name = Column(String)
while Pydantic models declare the types using :, the new type annotation syntax/type hints:
name: str
Have it in mind, so you don't get confused when using = and : with them.

from typing import List, Optional

from pydantic import BaseModel


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True        # <-------- これも大事

あー pydantic の方が書きやすいなぁ。

orm_mode は↓

But with ORM mode, as Pydantic itself will try to access the data it needs from attributes (instead of assuming a dict), you can declare the specific data you want to return and it will be able to go and get it, even from ORMs.
https://fastapi.tiangolo.com/tutorial/sql-databases/

Bigger Applications - Multiple Files¶

APIRouter

いまいちわからん・・

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}

Path operations with APIRouter¶

tags ってなんだっけ？？ path operationに使うものらしい。読み飛ばしたな

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}

Dependencies

これもよくわからん・・ これをデコレータで使うわけではなくて・・？ と思ったら次の項目見てわかった

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

Another module with APIRouter¶

router に dependencies 入れればいいのね。

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

routerには最後の / を入れないこと！

...the prefix must not include a final /.

The main FastAPI

で、こうやってつかうと

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

疲れちゃった。今日はここまで