API 코딩테스트용으로 fastapi를 사용해본 후 정리하는 글이다.

목차

1. FastAPI란?
2. FastAPI 특징: ASGI
3. 주요 기능별 코드 정리

1. FastAPI란?

FastAPI는 현대적이고, 빠르며(고성능), 파이썬 표준 타입 힌트에 기초한 Python3.6+의 API를 빌드하기 위한 웹 프레임워크이다.

2. FastAPI 특징: ASGI

FastAPI가 빠른 동작을 보여주는 이유 중 하나는 비동기식 웹 애플리케이션을 사용하기 때문이다.
기존의 WSGI(Web Server Gateway Interface)는 파이썬 웹 애플리케이션과 웹 서버 간의 인터페이스로 사용되었다. 하지만, 동기적인 프로그래밍 모델을 따르기 때문에 동시 처리에 제약이 있었다. Django, Flask같은 프레임워크가 WSGI 기반으로 동기 처리를 지원한다.
ASGI(Asynchronous Server Gateway Interface)는 비동기 웹 애플리케이션을 위한 파이썬 웹 서버와 웹 프레임워크 간의 인터페이스다. 비동기 처리와 동시 처리를 지원하여 더욱 향상된 성능을 보여준다. FastAPI는 ASGI를 기반으로 비동기 처리를 지원한다.

3. 주요 기능별 코드 정리

1) 설치

setup.sh와 같은 라이브러리 설치 파일에 추가한다.

pip install -r requirements.txt

requirements.txt에 라이브러리와 그 버전 정보를 추가한다.

fastapi==0.68.0
uvicorn==0.15.0

terminal에서 pip list를 하면 확인할 수 있다.

2) 데이터 불러오기

json 파일인 경우

파일 불러오고 쓸 함수를 하나씩 만들어놓고 시작하는 게 편하다.

# json 파일 읽기
def read_json_file(file_path):
    with open(file_path, "r") as file:
        data = json.load(file)
    return data
    
# json 파일 쓰기
def write_json_file(data, file_path):
	with open(file_path, 'w') as file:
    	json.dump(data, file)

그리고 각 파일들이 여러 문항에서 사용되기도 하기 때문에 이렇게 file path를 만들어 놓고 사용하였다.

users_file_path = os.path.join(os.path.dirname(__file__), "data/users.json")
items_file_path = os.path.join(os.path.dirname(__file__), "data/items.json")

2) API 만들기

import datetime
import json
import os

from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import JSONResponse
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

app = FastAPI()

GET

• 예시1

@app.get("/check")
def get_status():
	return {"message": "server is running"}

• 예시2

@app.get("/api/items/{item_id}")
def get_items(item_id:int):
	items_data = read_json_file(items_file_path)
    item = next((data for data in items_data if data["id"] == item_id), None)
    
    if user:
        return JsonResponse(content={"item": {"id": item["id"], "title": item["title"], "price": item["price"]}})
    else:
        raise HTTPException(detail="Item not found", status_code=404)

POST

type annotation을 활용해서 data validation해준다.

class Item(BaseModel):
	title: str
    price: int

@app.post("/api/item")
def register_item(data: Item):
    try:
        write_json_file(data, items_file_path)
        return JsonResponse(content={"message": "Item registered successfully"}, status_code=201)
    except:
        raise HTTPException(detail="Invalid data Format", status_code=400)

PUT, PATCH, DELETE도 사용할 수 있다.

3) 로그인 - 토큰 발급하기

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

def create_jwt_token(data: dict, expires_delta: datetime.timedelta):
	to_encode = data.copy()
    expire = datetime.datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

$\dots$

users_data = read_json_file(users_file_path)
user = next((data for data in users_data if data["username"] == username and data["password"] == password), None)
    
expires_delta = datetime.timedelta(minutes=60)
access_token = create_jwt_token(data={"sub": user["id"]}, expires_delta=expires_delta)

response_data = {
	"data": {"expire": int((datetime.datetime.utcnow() + expires_delta).timestamp() * 1000), "token": access_token}
    }
headers = {"Authorization": f"Bearer {access_token}"}
return JSONResponse(content=response_data, headers=headers)

이런 식으로 하면 access token을 생성할 수 있다.

4) 로그인 - 토큰 검증하기

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def validate_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if not user_id:
            raise HTTPException(detail="Invalid Token", status=401)
    except:
        raise HTTPException(
        detail="Could not validate credentials",
        headers={"WWW-Authorization": "Bearer"},
        status_code=401,
    )
    return user_id

@app.get("/api/users", response_model=User)
async def get_user(user: int = Depends(validate_token)):
    users_data = read_json_file(users_file_path)
    user = next((data for data in users_data if data["id"] == user_id, None)
    return user

5) 실행

host 또는 port가 정해져있을 때는 다음과 같이 실행한다.

uvicorn main:app --host 0.0.0.0 --port 7777 --reload

그리고 새로운 terminal을 열어 다음과 같이 테스트해볼 수 있다.

curl -X GET "0.0.0.0:7777/check" \
-H 'Content-Type: application/json'

curl -X POST "0.0.0.0:7777/api/item" \
-H 'Content-Type: application/json' \
-d '{
  "title": "shirts", 
  "price": 50000
  }'

---

참고
https://fastapi.tiangolo.com/ko/
https://fastapi.tiangolo.com/ko/tutorial/first-steps/
https://codechacha.com/ko/python-read-write-json-file/