FastAPI勉強メモ

# 参考資料
https://fastapi.tiangolo.com/

# 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.

```python
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](https://marketplace.visualstudio.com/items?itemName=ms-python.python)
- [pylance](https://marketplace.visualstudio.com/items?itemName=ms-python.vscode-pylance) : 静的型チェック、 コード入力補助機能を提供する
    - https://qiita.com/Tatejimaru137/items/94d84848d74c99b59157

# python typeを理解する
https://fastapi.tiangolo.com/python-types/

# Hello Worldを作ってみる
- main.pyファイルを作成して、以下のコードを書いて、保存する

```python
# 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を起動

```shell
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
```python
@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
```python
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
```python
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タイプを明示的にする
```python
@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
```python
# 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
```python
from typing import Optional

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

#### bool型のQuery parameter type
```python
# 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
```python
# 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
```python
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(...)
```python
@app.get("/items/")
async def read_items(q: str = Query(..., min_length=3)):
```

#### 複数値のQuery parameter
```python
# 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)):
```
```python
@app.get("/items/")
async def read_items(q: List[str] = Query(["foo", "bar"])):
```

#### Query parameter metadata
```python
@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
```python
# 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）
```python
@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, deprecated=True)):
```

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

```python
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を受け取ることができる

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

#### parameterの名前をjson keyにしたいことにembed=Trueを付ける
```python
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item = Body(..., embed=True)):
```
で以下のようなjson bodyを受け取ることができる

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


#### Multiple body parameters
```python
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を受け取ることができる

```json
// 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
```py
@app.put("/items/{item_id}")
async def update_item(
    item_id: int, importance: int = Body(...)
):
```
で以下のようなjson bodyを受け取ることができる

```json
{
    "importance": 5
}
```

#### json body fieldのvalidation & metadata
```python
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
```python
from typing import Set

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

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

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

#### array list
```python
class Image(BaseModel):
    url: HttpUrl
    name: str

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

```

#### 任意キーのjson body
```python
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
https://pydantic-docs.helpmanual.io/usage/types/#pydantic-types

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

```shell
pip3 install python-multipart
```

```python
from fastapi import FastAPI, Form

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

# Cookie parameters
```python
from fastapi import Cookie, FastAPI

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

# Header parameters
```python
from fastapi import FastAPI, Header

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

#### 重複するheader取得
```python
@app.get("/items/")
async def read_items(x_token: Optional[List[str]] = Header(None)):
```

# Response Model(serialization)
```python
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を指定する
```python
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を使う
```python
@app.get("/keyword-weights/", response_model=Dict[str, float])
```

#### 共通の項目をもつbaseクラスを作ることで重複を減らす
```python
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指定
```python
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に返す

```python
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ハンドラー
```python
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)
```python
@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
```python
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

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

#### 実行しますが、戻り値を使わないdependencyはdependenciesで渡す
```python
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
```python
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
```python
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
```

#### responseを返した後に、その他処理を行い時に、yieldを使う
```python
async def get_db():
    db = DBSession()
    try:
        yield db
        # responseを返した後の処理
    finally:
        db.close()
```

或いは[context manager](https://docs.python.org/3/reference/datamodel.html#context-managers)を使う

```python
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)

```python
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": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "alice@example.com",
        "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/

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

```python
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": "johndoe@example.com",
        "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を返す前に後の処理を行ったりすることができる

```python
@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の何が異なる場合

```python
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=["*"],
)
```

# 大きいプロジェクトの構成
![スクリーンショット 2020-12-08 22.53.51.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/114912/76b536c4-2eda-6867-6d6f-bf360b047741.png)

### APIRouter
app/routers/items.py

```python
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

```python
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

```python
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/

```python
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(静的ファイル)
```shell
pip install aiofiles
```

```python
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

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

# testing
```shell
pip install pytest
pip install requests
```

```python
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を返す１
```python
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が自動で生成される

```python
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を返す
```python
@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を生成
https://fastapi.tiangolo.com/advanced/additional-responses/

```python
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できます。

```python
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を起動
```