프로토타입 개발 : Voila, Streamlit
프론트 : Streamlit
백엔드 : Streamlit -> FastAPI로 대체
서비스로 만들 땐 Streamlit 이슈가 있음
-> 회사 입사 시 어떤식으로 서버 구축돼있는지 확인 필요
하나의 큰 아키텍쳐 : 모놀리식 아키텍쳐
- 문제점 : 전체 서버가 전부 배포가 되는거라 속도가 빨라질 수 없음(다른 문제도 있음)
- 해결 : 따라서 ,서버를 나눔 - 마이크로서비스 아키텍쳐(MSA)
IP란 인터넷에 연결되어 있는 모든 장치들(컴퓨터, 서버 장비, 스마트폰 등)을 식별할 수 있도록 각각의 장비에게 부여되는 고유 주소이다
https://study-recording.tistory.com/13
위 사이트 참고해서 정리
- 정보를 주고 받을 때 지켜야 하는 통신 프로토콜(규약), 약속
기본적으로 80번 포트 사용, 그 외에는 사용불가
- HTTP Method 종류
- GET : 정보 요청 (READ)
- POST : 정보 입력 (Create)
- PUT : 정보 업데이트 (Update)
- PATCH : 정보 업데이트 (Update)
- DELETE : 정보 삭제 (Delete)
웹에서 GET Method 사용해서 데이터 전송 가능
ID가 402인 사용자 정보 가져올 때 아래와 같이 작성 !
- Path Parameter 방식 : /users/402
- Query Parameter 방식 : /users?id=402
- /users/kyle : Path Parameter
- /users?name=kyle : Query Parameter
- 언제 어떤 방식을 사용해야 할까?
- Resource 식별 시 : Path Parameter 적합
- Query 식별 시 : Query Parameter 적합
- Get Method : 정보 READ 위해
- FastAPI는 데코레이터로 GET, POST 표시 !!!!
- @app.get @app.post
from fastapi import FastAPI
import uvicorn
app = FastAPI()
@app.get("/users/{user_id}")
def get_user(user_id):
return {"user_id": user_id}
if __name__ == '__main__':
uvicorn.run(app, host = "0.0.0.0", port = 8000)
from fastapi import FastAPI
import uvicorn
app = FastAPI()
fake_items_db = [{"item_name" : "Foo"}, {"item_name" : "FIFA2023"}, {"item_name" : "LOL"}]
@app.get("/items/")
def read_item(skip: int = 0, limit: int = 10):
return fake_items_db[skip : skip + limit]
if __name__ == '__main__':
uvicorn.run(app, host = "0.0.0.0", port = 8000)
from typing import Optional
from fastapi import FastAPI
import uvicorn
app = FastAPI()
fake_items_db = [{"item_name" : "Foo"}, {"item_name" : "FIFA2023"}, {"item_name" : "LOL"}]
@app.get("/items/{item_id}")
def read_item(item_id: str, q: Optional[str] = None):
if q:
return {"item_id": item_id , "q": q}
return {"item_id": item_id}
if __name__ == '__main__':
uvicorn.run(app, host = "0.0.0.0", port = 8000)
- 특정 파라미터는 Optional으로 하고 싶은 경우
- typing 모듈의 Optional 사용
- Optional 사용해 이 파라미터는 Optional임을 명시(기본 값 None)
- 클라이언트에서 API에 데이터 보낼 때, Request Body 사용
- 클라이언트 → API: Request Body
- API의 Response → 클라이언트 : Response Body
- Request Body에 데이터 보내고 싶다면 POST Method 사용
- GET Method는 URL, Request Header로 데이터 전달
- POST Method는 Request Body에 데이터 넣어 보냄
- Body의 데이터 설명하는 Content-Type이란 Header 필드 존재하고, 어떤 데이터 타입인지 명시해야 함
from typing import Optional
from fastapi import FastAPI
import uvicorn
# pydantic으로 Request Body Data 정의 !!!
from pydantic import BaseModel
# name, price -> 필수 !!
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None
app = FastAPI()
# Post 요청으로 Item 생성하는 예제(Create)
# Type Hinting에 위에서 생성한 Class 주입
# Request Body Data를 Validaion
@app.post("/items/")
def create_item(item: Item):
return item
if __name__ == '__main__':
uvicorn.run(app, host = "0.0.0.0", port = 8000)
- Scheamas에서 pydantic으로 정의한 내용 볼 수 있음
Swagger UI에서 Data Validation 실습
Float Type에 String 넣어서 Error 발생시켜보기
- Server Response에서 Error 발생 알림 (Validation Check!!)
from typing import Optional
from fastapi import FastAPI
import uvicorn
from pydantic import BaseModel
# Request Data와 Response Data 다르게 할 수 있기 때문에 아래와 같이 클래스 정의
class ItemIn(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None
class ItemOut(BaseModel):
name: str
price: float
tax: Optional[float] = None
app = FastAPI()
@app.post("/items/", response_model = ItemOut)
def create_item(item: ItemIn):
return item
if __name__ == '__main__':
uvicorn.run(app, host = "0.0.0.0", port = 8000)
- 역할
- OIutput Data를 해당 정의에 맞게 변형
- 데이터 Validation
- Response에 대한 Json Schema 추가
- 자동으로 문서화
from fastapi import FastAPI, Form, Request
from fastapi.templating import Jinja2Templates
import uvicorn
# 정상 Code
app = FastAPI()
templates = Jinja2Templates(directory = './')
# 제출을 누르면 login 함수가 실행됨(POST 요청)
# Request 객체로 Request 받음
@app.get("/login/")
def get_login_form(request: Request):
return templates.TemplateResponse('login_form.html', context = {"request" : request})
# Form(...) : Python ellipsis : Required(꼭 필수 요소)의미
@app.post("/login/")
def login(username: str = Form(...), password: str = Form(...)):
return {"username": username}
if __name__ == "__main__":
uvicorn.run(app, host = "0.0.0.0", port = 8000)
- 파이썬에서 사용할 수 있느 템플릿 엔진 : Jinja Template (Front-End 구성)
- templates.TemplateResponse로 해당 HTML로 데이터 보냄
<!DOCTYPE html>
<html lang = "en">
<head >
<meta charset="UTF-8">
<title>Sample Login Form</title>
</head>
<body>
<form method = "post">
<input type ="string" name="username" value="{{username}}"/>
<input type ="password" name="password" value="{{password}}"/>
<input type ="submit">
</form>
</body>
</html>
# Error Code
app = FastAPI()
@app.post("/login/")
def login(username: str = Form(...), password: str = Form(...)):
return {"username": username}
if __name__ == "__main__":
uvicorn.run(app, host = "0.0.0.0", port = 8000)
- Error Code -> "GET/login/HTTP/1.1" : 405 Method Not Allowed
- URL 접속이 Get Method 요청된 것인데 없어서 이런 에러 발생
- Validation Check Logic
- 조건 1: 올바른 url 입력 받음(url)
- 조건 2: 1~10 사이의 정수 입력 받음(rate)
- 조건 3: 올바른 폴더 이름을 입력 받음(target_dir)
- 사용할 수 있는 방법
- 일반 Python Class 활용한 Input Definition 및 Validation
- Dataclass를 (Python 3.7이상) 활용한 Input Definition 및 Validaiton
- Pydantic 활용한 Input Definition 및 Validation
from pydantic import BaseModel, HttpUrl, Field, DirectoryPath
class ModelInput03(BaseModel):
url: HttpUrl
rate: int = Field(ge=1, le=10)
target_dir: DirectoryPath
애플리케이션은 종종 설정을 상수로 코드에 저장
이것은 Tweleve-Factor 위반!
Twelve-Facotr App은 설정을 환경 변수(envvars, env)에 저장함
환경 변수는 코드 변경 없이 쉽게 배포때마다 쉽게 변경 가능
❌ 좀 더 많은 복습 필요함 ! 다시 정리하러 옴
Event 발생 시, 그 처리 담당 함수
FastAPI에선 Application실행 시 종료될 때 특정함수 실행
Slack 알리기 가능
Ex) startup ML Model Load, shutdown 시 로그 저장, 중간에 저장안됐지만 메모리에 잇는것도 저장가능
from fastapi import FastAPI
import uvicorn
app = FastAPI()
items = {}
@app.on_event("startup")
def startup_event():
print("Start Up Event")
items["foo"] = {"name" : "Fighters"}
items["bar"] = {"name" : "Tenders"}
@app.on_event("shutdown")
def shutdown_event():
print("Shutdown Event!")
with open("log.txt", mode = "a") as log:
log.write("Application shutdown")
@app.get("/items/{item_id}")
def read_items(item_id: str):
return items[item_id]
if __name__ == '__main__':
uvicorn.run(app, host = "0.0.0.0", port = 8000)
- API Router는 더 큰 애플리케이션들에서 많이 사용되는 기능
- API Endpoint를 정의
- Python Subpackage
- APRouter는 Mini FastAPI로 여러 API 연결해서 활용
- 기존에 사용하던 @app.get, @app.post 사용하지 않고, router 파일을 따로 설정하고 app에 import해서 사용
from fastapi import FastAPI, APIRouter
import uvicorn
user_router = APIRouter(prefix = "/users")
order_router = APIRouter(prefix = "/orders")
@user_router.get("/", tags = ["users"])
def read_users():
return [{"username" : "Rick"}, {"username" : "Morty"}]
@user_router.get("/me", tags = ["users"])
def read_user_me():
return {"username": "fakecurrentuser"}
@user_router.get("/{username", tags=["users"])
def read_user(username: str):
return {"username": username}
@order_router.get("/", tags = ["orders"])
def read_orders():
return [{"order" : "Taco"}, {"order" : "Burritto"}]
@order_router.get("/me", tags = ["orders"])
def read_order_me():
return {"my_order": "taco"}
@order_router.get("/{order_id}", tags = ["orders"])
def read_order_id(order_id : str):
return {"order_id" : order_id}
app = FastAPI()
if __name__ == '__main__':
app.include_router(user_router)
app.include_router(order_router)
uvicorn.run(app, host = "0.0.0.0", port = 8000)
- Error Handling은 웹 서버를 안정적으로 운영하기 위해 반드시 필요한 존재 !!
- 서버에서 Error가 발생한 경우, 어떤 Error가 발생했는지 알아야 하고, 요청한 클라이언트에 해당 정보를 전달해 대응할 수 있어야 함
- 서버 개발자는 모니터링 도구를 사용해 Error Log를 수집해야 함
- 발생하고 있는 오류를 빠르게 수정할 수 있도록 예외 처리를 잘 만들 필요가 있음
from fastapi import FastAPI
import uvicorn
app = FastAPI()
items = {
1: "BoostCamp",
2: "AI",
3: "Tech"
}
@app.get("/v1/{item_id}")
async def find_by_id(item_id: int):
return items[item_id]
if __name__ == '__main__':
uvicorn.run(app, host = "0.0.0.0", port = 8000)
- 설정 Key값 이외의 값을 넣었을 때 Internal Server Error 500 Return
- 이럴 경우, 클라이언트는 어떤 에러가 떴는지? 자세한 에러 알려면 서버에 직접 접근해서 LOG 확인해야함...;;;;
- Error Handling 더 잘하려면 아래 사항을 클라이언트에 전달하도록 코드 짜야함
- Error Message
- Error 이유
from fastapi import FastAPI, HTTPException
import uvicorn
app = FastAPI()
items = {
1: "BoostCamp",
2: "AI",
3: "Tech"
}
@app.get("/v1/{item_id}")
async def find_by_id(item_id: int):
return items[item_id]
# f string이였네
@app.get("/v2/{item_id}")
async def find_by_id(item_id: int):
try:
item = items[item_id]
except KeyError:
raise HTTPException(status_code = 404, detail=f"아이템을 찾을 수 없습니다 [id: {item_id}]")
return item
if __name__ == '__main__':
uvicorn.run(app, host = "0.0.0.0", port = 8000)
- 프로세스 외에서 백그라운드로 돌리고 나중에 결과 받아도 좋으면 사용함
- FastAPI는 Starlett이라는 비동기 프레임워크를 래핑해서 사용
- FastAPI의 기능 중 Background Tasks 기능은 오래 걸리는 작업들을 backgroud에서 실행함
- Online Serving에서 CPU 사용이 많은 작업들을 Backgroud Task로 사용하면, 클라이언트는 작업완료 기다리지 않고 즉시 Response 받는다?
- 즉시 Response 받는 게 뭘까? 아...비동기처럼 백그라운드 하나 돌리고, 클라이언트가 요청한 건 또 비동기의 특징처럼 병렬적으로 처리해줌
- 아하 !
- 작업 결과물을 조회할 때는 Task를 어딘가에 저장해두고, GET 요청을 통해 Task가 완료됐는지 확인
from fastapi import FastAPI
import uvicorn
# 앱 생성
app_2 = FastAPI()
# 비동기 작업이 등록됐을 때, HTTP Response 202(Accepted!)를 보통 리턴함 !!
@app_2.post("/task", status_code = 202)
async def create_task_in_background(task_input: TaskInput, background_tasks: BackgroundTasks):
backgroun_tasks.add_task(cpu_bound_task, task_input.wait_time)
return "ok"
start_time = datetime.now()
run_tasks_in_fastapi(app_2, tasks)
end_time = datetime.now()
print(f"Background Tasks: Took {(end_time - start_time).seconds}")
- 아예 처음 진행 시에는 일단 하나의 파일에 모두 코드 작성해서 FastAPI에 익숙해지기
- 그 이후에 다른 프로젝트 구조를 참고해서 코드 개선
- 클린 아키텍처 류의 책을 보면서 계속 구조 고민하기
- 현재 가지고 있는 코드를 Class로 변경해보기
- Pydantic Use Case 탐색해보기
- 목표 설정 -> 기능 정의 -> 하나씩 구현
- 프로젝트를 FastAPI 백엔드로 구현 ! 목표설정