이 글은 FastAPI, Uvicorn, Websocket 기반 Open API 서버 간단히 개발하는 방법을 정리한다. FastAPI를 이용하면, 매우 쉽게 Open API 서버를 개발할 수 있다.
FastAPI는 비동기 API 서버를 지원하며, uvicorn과 같은 ASGI 서버를 사용하여 실행할 수 있다. 이를 통해 빠른 성능과 비동기 처리를 구현할 수 있다. 자동 대화형 API 문서도 제공되어 개발자가 API를 쉽게 이해하고 사용할 수 있다.
FastAPI는 다음 웹 어플리케이션 프레임웍인 Flask, Django와 함께 활용하면 좋다.
패키지 설치는 다음과 같다.
pip install fastapi aiohttp uvicorn
서버 개발
크게 2개 유형의 API를 개발한다. 하나는 오래 걸리는 계산 함수, 다른 하나는 대용량 데이터 파일 전달이다. 대용량 파일은 청크로 분할해 전달한다. server.py 코드는 다음과 같다.
import json, time, logging
from fastapi import FastAPI, BackgroundTasks, WebSocket
from fastapi.middleware.cors import CORSMiddleware
# Set up logging to debug
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# uvicorn open_api_server:app --reload --port 8001 --ws-max-size 16777216 # https://www.uvicorn.org
app = FastAPI()
# CORS middleware. considering security
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allows all origins
allow_credentials=True,
allow_methods=["*"], # Allows all methods
allow_headers=["*"], # Allows all headers
)
def calculate_dataset(length):
logger.debug('Calculation started...')
time.sleep(30) # 30 seconds
# TODO call your calculation functions
data = 3.14
logger.debug(f'End')
return data
@app.post("/v1/calc/dataset")
async def calculate_dataset(background_tasks: BackgroundTasks, length: str):
logger.debug('Calculation started...')
results = calculate_dataset(length)
return {"results": results}
@app.websocket("/ws/dataset") # don't remove /ws prefix to use websocket
async def websocket_endpoint(websocket: WebSocket):
logger.debug('Trying to connect...')
await websocket.accept()
logger.debug('Connection accepted.')
with open('output_xml.json') as json_file:
data = json.load(json_file)
data = json.dumps(data)
logger.debug('Message length: ' + str(len(data)))
CHUNK_SIZE = 64 * 1024 # 64KB
for i in range(0, len(data), CHUNK_SIZE):
chunk = data[i:i+CHUNK_SIZE]
print(f'chunk: {i}')
await websocket.send_text(chunk)
logger.debug('JSON data sent.')
다음과 같이 실행한다.
uvicorn open_api_server:app --reload --port 8001
클라이언트 개발
대용량 데이터와 결과를 클라이언트에서 처리할 때는 타임아웃 설정과 청크 다운로드 루프를 구현해야 한다. client.py를 코딩한다.
import json, traceback, asyncio, websockets, aiohttp # import httpx
from aiohttp import ClientSession, ClientTimeout
async def call_calc_dataset():
params = {"length": "10"}
t = ClientTimeout(total=60*2) # 2 minutes
async with aiohttp.ClientSession(timeout=t) as session:
async with session.post('http://localhost:8001/v1/calc/dataset', params=params) as resp:
results = await resp.text()
print(results)
async def connect():
async with websockets.connect('ws://localhost:8001/ws/dataset') as websocket:
CHUNK_SIZE = 64 * 1024 # 64KB
full_data = None
received_count = 0
try:
while True:
chunk = await asyncio.wait_for(websocket.recv(), timeout=5) # adjust timeout value considering internet speed
if full_data is None:
full_data = chunk
else:
full_data += chunk
received_count += len(chunk)
print('Received data: ', received_count)
except asyncio.TimeoutError:
print("Timeout error: The server didn't respond within 5 seconds")
except Exception as e:
print(e)
pass
print('Received data: ', received_count)
data = json.loads(full_data)
try:
os.remove('big_xml.json')
except:
pass
with open('big_xml.json', 'w') as json_file:
json.dump(data, json_file)
print('JSON data saved to file.')
def main():
try:
loop = asyncio.get_event_loop()
loop.run_until_complete(call_calc_dataset())
loop.run_until_complete(connect())
except Exception as e:
print(traceback.format_exc())
if __name__ == '__main__':
main()
다음과 같이 실행한다.
python client.py
서버와 클라이언트에서 실행 결과는 다음과 같다. 다음과 같으면 정상 동작되는 것이다.
마무리
FastAPI는 Sebastián Ramírez라는 개발자에 의해 처음 개발되었다. 그는 현재 독일 베를린에 살고 있고, 많은 오픈소스 기여활동으로 유명하다.
댓글 없음:
댓글 쓰기