[FastAPI] Docs Bookmark - Tutorial 3

python 3.6 ~ python 3.8

Fastapi 0.95.0

---

Dependencies

⌾ What is "Dependency Injection"

⌵ 코드의 작동 및 사용에 필요한 의존성을 선언할 방법이 있음을 의미

⌵ 반복적인 logic, DB connection 공유, 보안|인증|역할 요건 등에 유용하다

First Steps

⌾ Create a dependency or dependable & Import Depends 

⌾ Declare the dependency, in the "dependant"

• Depends에는 single 파라미터만 지정하며 해당 파라미터는 함수와 유사하다

...
from fastapi import Depends, FastAPI

async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

 

새로운 Request가 올때마다 FastAPI는:

• Dependency 함수 호출
• 해당 함수로부터 결과 도출
• path operation 함수의 파라미터에 해당 결과 할당

 

⌾ Share Annotated dependencies

...
CommonsDep = Annotated[dict, Depends(common_parameters)]

@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons

@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons

⌾ To async or not to async

Classes as Dependencies

⌾ What makes a dependency

⌵ dependency는 함수로 선언하는 것만이 유일한 방법은 아니다

⌵ 중요한 것은 'callable'해야 한다는 것

⌵ 따라서 개체가 있으면 함수가 아니어도 호출할 수 있다

something()
#or
something(some_argument, some_keyword_argument="foo")

⌾ Classes as dependencies

• __init__에 파라미터 값 정의

...
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

class CommonQueryParams:
    def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

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

⌾ Shortcut

async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):

↓ ↓

async def read_items(commons: Annotated[CommonQueryParams, Depends()]):

Sub dependencies

• First dependency "dependable" query_extractor
• Second dependency, "dependable" and "dependable" query_or_cookie_extractor
  
  • dependency 함수에서 다른 dependency도 선언
  • 해당 함수의 q 파라미터는 query_extractor에서 반환된 값을 할당
• Use the dependency async def read_query

...
def query_extractor(q: Union[str, None] = None):
    return q

def query_or_cookie_extractor(
    q: Annotated[str, Depends(query_extractor)],
    last_query: Annotated[Union[str, None], Cookie()] = None,
):
    if not q:
        return last_query
    return q

@app.get("/items/")
async def read_query(
    query_or_default: Annotated[str, Depends(query_or_cookie_extractor)]
):
    return {"q_or_cookie": query_or_default}

⌾ Using the same dependency multiple times

⌵ 여러 dependency에 공통 sub-dependency가 있는 예시와 같이, 동일 path operation에서 dependency 하나가 여러 번 선언된 경우 FastAPI는 한 request당 한 번만 sub-dependency를 호출한다

⌵ sub-dependency를 여러 번 호출하는 대신, 호출한 결과물을 cache에 저장하여 필요한 모든 dependants에게 전달한다

⌵ 모든 단계에서 dependency를 호출하고 싶다면:

async def needy_dependency(fresh_value: Annotated[str, Depends(get_value, use_cache=False)]):
    return {"fresh_value": fresh_value}

Dependencies in path operation decorators

⌵ dependency의 반환값은 필요없지만(혹은 반환값이 없거나) dependency는 필요한 경우,

⌵ Depends 대신 dependencies 리스트를 추가해 사용할 수 있다

⌾ Add dependencies to the path operation decorator 

⌾ Dependencies errors and return values

from fastapi import Depends, FastAPI, Header, HTTPException
from typing_extensions import Annotated

app = FastAPI()

async def verify_token(x_token: Annotated[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: Annotated[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"}]

일반적인 dependency와 동일하게 실행되지만, 반환값이 path operation function에 전달되지 않는다

Dependencies in path operation decorators

⌵ Application의 모든 path operation에 적용

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

⌾ A database dependency with yield

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

⌾ Sub-dependencies with yield

...
async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()

async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)

async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

⌾ Dependencies with yield and HTTPException

⌾ Context Managers

• python-With Statement Context Managers
• python-Reading and Writing Files

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

---

Security

Intro

⌾ OAuth2 

⌵ 인증 및 권한 부여를 처리하는 몇 가지 방법을 정의하는 규격

⌵ third party를 사용하여 인증하는 방법 포함 (Google, Facebook, Github, ...)

⌾ OAuth1 

⌾ OpenAPI

OpenAPI에서 정의하는 Security Schemes:

• api-key
  
• http
• oauth2
• openIdConnect

First Steps

pip install python-multipart

⌾ OAuth2PasswordBearer

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
from typing_extensions import Annotated

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/items/")
async def read_items(token: Annotated[str, Depends(oauth2_scheme)]):
    return {"token": token}

docs에 'Authorize' 버튼이 추가된다

⌾ password flow

Get Current User

• Create a user model  User
• Create a get_current_user dependency  get_current_user
• Get the user get_current_user
• Inject the current user  read_users_me

...
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


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

def fake_decode_token(token):
    return User(
        username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
    )

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    user = fake_decode_token(token)
    return user

@app.get("/users/me")
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

Simple OAuth2 with Password and Bearer

⌾ Get the username and password

• OAuth2의 password flow를 이용중일 때, 
• 필드명은 username, password (user-name이나 email은 작동하지 않음)
• form data로 전송

⌾ scope

• 공백이 없는 문자열
• 보통 특정 보안 권한을 선언할 때 사용한다:

⌾ Code to get the username and password

OAuth2PasswordRequestForm

• 다음과 같은 form body를 선언하는 dependency
• username, password, optional scope field, optional grant_type
• optional client_id, client_secret

...
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

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


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = 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

# Sub dependency
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

# Update dependency, 사용자가 활성화된 경우
async def get_current_active_user(
    current_user: Annotated[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: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user_dict = fake_users_db.get(form_data.username)
    # DB에 해당 사용자가 없으면, 오류 반환
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    # DB에서 찾은 사용자의 Hashed password와 hashing 작업을 거친 입력받은 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: Annotated[User, Depends(get_current_active_user)]
):
    return current_user

OAuth2 with Password (and hashing), Bearer with JWT tokens

⌾ About JWT

• "Json Web Tokens"
• jwt.io

pip install "python-jose[cryptography]"

⌾ Password hashing

pip install "passlib[bcrypt]"

⌾ Hash and verify the passwords

 

from datetime import datetime, timedelta
from typing import Union

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
from typing_extensions import Annotated

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" #import secrets
ALGORITHM = "HS256" #JWT 알고리즘
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: Union[str, None] = None


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


class UserInDB(User):
    hashed_password: str


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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()


# password 검증
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


# hashing
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)


# 비밀번호 검증 후, user 반환
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: Union[timedelta, None] = 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: Annotated[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: Annotated[User, Depends(get_current_user)]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


# Token 만료 시간 (timedelta)
@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[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
    )
    return {"access_token": access_token, "token_type": "bearer"}


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


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

---

Middleware

API 호출 전후로 작업을 처리하는 기능

https://blog.neonkid.xyz/271

⌾ Create a middleware

middleware receives:

• request
• request를 파라미터로 수신할 call_next 함수
• response 반환 전, 추가로 수정할 수 있음

@app.middleware("http")

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

---

CORS (Cross-Origin Resource Sharing)

• 웹 애플리케이션이 자신과 다른 Origin(domain, protocol, port)에서 리소스를 요청할 때 
• 추가 HTTP 헤더를 사용하여, 한 Origin에서 실행중인 웹 애플리케이션이 다른 Origin의 자원에 접근할 수 있는 권한을 부여하도록 브라우저에 알려주는 체제
• CORS

⌾ Use CORSMiddleware

백엔드 허용 여부 지정:

• Credentials (Authorization headers, Cookies, ...)
• 특정 HTTP method (또는 wildcard '*')
• 특정 HTTP header (또는 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"

arguments:

• allow_origins : cross-origin request를 허용하는 origin 목록
• allow_origin_regex : cross-origin request 허용을 위한 Origin 정규표현식
• allow_methods : 지원 가능한 HTTP Method
• allow_headers : 지원 가능한 HTTP header 목록 
• allow_credentials : cross-origin request에 대한 cookie 지원 여부 (default false)
• expose_headers : 브라우저에서 access 할 수 있는 response header
• max_age : 브라우저가 CORS response를 cache 할 최대 시간(s) (default 600)

---

SQL(relational) Database

pip install sqlalchemy

⌾ Create the SQLAlchemy parts

## Import the SQLAlchemy parts
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

## Create a database URL for SQLAlchemy
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

## Create the SQLAlchemy engine
## connect_args={"check_same_thread": False}는 SQLite에만 필요
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

 

⌾ Create data

• add
• commit
• refresh

⌾ ...

---

Bigger Applications - Multiple Files

---

Background Tasks

response 반환 후 실행할 background task 정의

⌾ Using BackgroundTasks

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

⌾ Caveat

---

Metadata and Docs URL

⌾ Metadata for API

Parameter	Type	Description
title	str	The title of the API
description	str	A short description of the API. It can use markdown
version	string	The version of the API. This is the version of your own application, not of OpenAPI.
terms_of_service	str	A URL to the Terms of Service for the API. If provided, this has to be a URL.
contact	dict	The contact information for the exposed API. It can contain several fields.
licence_info	dict	The license information for the exposed API. It can contain several fields.

⌾ Metadata for tags

• name(required) : tags parameter와 동일한 tag 이름을 가진 str
• description : tag를 위한 간단한 설명이 있는 str (markdown 가능)
• externalDocs : 외부 문서 설명
  • description
  • url(required)

⌾ OpenAPI URL

• Default OpenAPI schema /openai.json
• openai_url 파라미터를 통해 구성할 수 있다
• openai_url=None으로 설정하면 비활성화 할 수 있다

from fastapi import FastAPI

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

• Swagger UI (/docs)
  - docs_url
  - docs_url=None, SwaggerUI 비활성화
• ReDoc (/redoc)
  - redoc_url
  - redoc_url=None, ReDoc 비활성화

---

Static Files

StaticFiles를 통해 자동으로 디렉토리에서 static file을 제공할 수 있다

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

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

---

Testing (sync)

pip install httpx
pip install pytest

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


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

# TestClient 생성
client = TestClient(app)

# test_로 시작하는 함수 생성
def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

---

Debugging

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

---

 

FastAPI

[FastAPI] 10. Middleware를 이용한 전후 처리