회사에서 FAST API를 사용하게 되서, 작업을 시작했습니다.
Node.js와 다르게 몇몇 다른 것들이 보여 정리 할려고합니다.
요즘 ChatGPT가 유행이라, FAST API에 관련된 것들을 물어봤는데 정리를 잘해서 알려줘서 너무 좋았습니다.
Basemodel에 대한 chatGPT의 설명입니다.
from pydantic import BaseModel
class ResponseModel(BaseModel):
error: str = Field(title="error")
message: str = Field(title="message")
timestamp: int = Field(title="time")
data: Union[list, dict] = Field(title="data")
요런식으로 사용해서, BaseModel을 통하여 유효성 검사를 할 수 있습니다.
덧붙여 Field에 대한 설명입니다.
Field
Union
유효성 검사를 위해 사용합니다. 특정한 타입의 데이터가 들어오는지 체크합니다.
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = Query(default=100, lt=1000)):
items = {"skip": skip, "limit": limit}
return items
http://localhost:8000/items/?skip=0&limit=10 을 보냈을 경우, limit의 1000보다 작고 값이 기본값이 100입니다. limit가 1000보다 크면 에러를 반환합니다
Node.js의 경우는 app.js에 입구에서 ('/')로 들어와 router 폴더안에 router.js파일에서 분기하여 각 user.js 안에서 createuser, deleteuser 하는 식으로 구성했었습니다.
이것과 비슷하게 APIRouter를 통해서 해결할 수 있습니다. 대부분 main.py에 include_router를 집어넣던데 위에 방법이 더 좋다고 생각합니다
그렇기 위해서는init.py을 생성해야합니다.
from fastapi import FastAPI
from router import router
app = FastAPI()
app.include_router(router)
@app.get("/")
async def hello():
return {"message": "Hello Applications!"}
from fastapi import APIRouter
from .user import user
router = APIRouter()
router.include_router(user)
from fastapi import APIRouter
user = APIRouter(
prefix='/user',
tags=['User'],
)
async def request(client, url):
response = await client.get(url)
return response.json()
@user.get('/', tags=['User'], summary="모든 사용자")
async def get_user():
return '모든 유저'
@user.get('/{user_id}', tags=['User'], summary="특정 사용자")
async def get_user_id():
return '특정 유저'
이런식으로하면 분기가 가능합니다. init.py에서 여러가지로 분기하고 한 파일에 특정 query를 다 모을 수 있을 것입니다.
chatGPT는 요런 방식을 쓰던데, 별로라고 생각했습니다.
이제는 분기를 했으니 http 요청을 해볼려고합니다. FAST API에서는 비동기방식을 사용하게 되었습니다.
파이썬 사용자라면 http 요청으로 익숙한 requests를 사용할 것입니다.
하지만 FAST API는 비동기 기반이므로 httpx
와 aiohttp
를 사용해야합니다.
실제로 여러 코드를 보내면 requests의 경우 응답을 받는게 보장되지않음을 확인할 수 있습니다.
Node.js의 경우 Promise.all로 보장받을 수 있지만 fastapi에는 다음과 같게 사용합니다.
async def make_requests():
async with httpx.AsyncClient() as client:
response_1 = client.get('https://httpbin.org/get')
response_2 = client.post('https://httpbin.org/post', json={'key': 'value'})
response_3 = client.get('https://httpbin.org/redirect-to?url=https%3A%2F%2Fwww.google.com%2F')
response_4 = client.get('https://httpbin.org/status/404')
responses = await asyncio.gather(response_1, response_2, response_3, response_4)
return responses
asyncio.gather()를 사용해서 요청의 결과를 모아서 처리할 수 있습니다.
그럼에도 불구하고 requests를 사용해야하는 경우 thread를 늘려서 동기적방식을 사용할 수 있습니다.
with ThreadPoolExecutor(max_workers=3) as pool:
response_list = list(pool.map(get_url,url_list))
위 부분에 대해 공부하면서 async를 사용하면 uvicorn에 올라가서 이벤트루프를 사용하는 걸 알게되었습니다.
파이썬에서 async def 위에 보면 @가 보입니다. 이건 데코레이터 함수입니다.
def my_decorator(func):
def wrapper(*args, **kwargs):
print("Before the function is called.")
result = func(*args, **kwargs)
print("After the function is called.")
return result
return wrapper
@my_decorator
def say_hello(name):
print(f"Hello, {name}!")
say_hello("Alice")
데코레티어는 함수 또는 클래스의 기능을 확장하거나 변경하고자 할 때 사용합니다.
from fastapi import Depends, FastAPI
app = FastAPI()
async def properties(offset: int = 0, limit: int = 100):
return {"offset": offset, "limit": limit}
@app.get("/books/")
async def get_books(params: dict = Depends(properties)):
return params
의존성 주입이라는 단어로 잘 사용됩니다. Depends를 설정하면, 위의 요청으로 왔을 때 Depends를 먼저 실행시킨 후, 해당 요청을 실행시킵니다.
위의 설명으로는 왜 굳이 사용하지 싶을테지만, db 연결하고 연결을 끊고, 보안검사를 실행한다던가 할때 , try 문 안에 넣어서 실행단계 안으로 넣지않음으로 더 깔끔한 코드 작성이 가능합니다.
코드 실행에 있어 선행적으로 검사하거나 실행되기를 원할 때 사용을 위해 있습니다. 그런의미에서 '의존성 주입' 입니다.