안녕하세요, 홉스의 안기욱입니다.
오늘은 저희의 백엔드 서버 인프라를 Flask에서 FastAPI로 전환한 후기에 대해서 공유드리려고 합니다. 이 글에서는 Flask에서 FastAPI로 서버를 전환하게 된 이유와 전환을 위한 가이드라인, 그리고 추가적인 FastAPI 사용 팁과 덩달아 사용하게 된 비동기 라이브러리에 대해 소개합니다. 마지막으로 전환에 대한 간단한 제 생각도 같이 공유드릴 예정입니다.
시작하기에 앞서, 이 글은 2021년 10월 2일 PyCon Korea 2021에서 발표한 대본을 토대로 작성하였음을 밝힙니다.
목차를 보시려면 펼쳐주세요.
Flask를 FastAPI로 전환하게 된 이유
Flask 기반의 백엔드 프로젝트 작업을 진행하다가, 웹소켓 엔드포인트를 추가해야 하는 요구사항이 생겼습니다. 처음에는 node 기반의 백엔드 서버를 따로 띄우려는 기획을 했었으나, 빠른 제품 개발을 위해 기존에 Python으로 구현해둔 모델 등의 여러 구현체들을 활용하고 싶었습니다. 그렇지만 웹소켓은 필연적으로 통신 대기 시간이 많이 발생하므로, asyncio를 활용해서 비동기로 동작하는 엔드포인트를 만들고 싶었습니다.
Flask는 2021년 5월 21일에 런칭한 2.0.0 버전에서 asyncio를 이용한 엔드포인트 지원을 추가하였지만, 웹소켓 엔드포인트를 만들어야 했던 3월 당시에는 그러한 방법이 없었습니다. 만약 전통적인 방법으로 웹소켓 엔드포인트를 구현할 경우, 통신 대기 시간의 유휴 자원이 낭비되므로 이러한 방법을 선택할 수는 없었습니다. 이에 코드베이스를 유지하면서 asyncio 엔드포인트를 작성할 수 있는 방법을 찾기 시작했습니다.
asyncio 간단한 소개
여기서 잠깐 asyncio에 대한 간단한 설명을 드리도록 하겠습니다.
전통적인 Python 코드는 파일을 읽거나 쓰는 작업, 네트워크 통신 작업 등 I/O 작업을 수행하면 필연적으로 대기 시간이 발생하게 됩니다. 그래서 보통 이러한 작업들을 구현할 때에는 Celery를 쓰거나 이러한 대기 시간을 마냥 기다리다가 다음 작업을 실행하는 블로킹 방식으로 구현을 하게 되는데요.
asyncio는 이벤트 루프를 사용해 이런 작업들을 관리하여 대기 시간을 낭비하지 않고 다른 작업을 실행할 수 있도록 해 줍니다. 이벤트 루프는 asyncio의 가장 중요한 부분이지만 일반적인 사용에서는 자세한 디테일을 몰라도 괜찮습니다. 다음은 공식 문서의 머릿말 내용입니다.
이벤트 루프는 모든 asyncio 응용 프로그램의 핵심입니다. 이벤트 루프는 비동기 태스크 및 콜백을 실행하고 네트워크 IO 연산을 수행하며 자식 프로세스를 실행합니다.
응용 프로그램 개발자는 일반적으로 asyncio.run()과 같은 고수준의 asyncio 함수를 사용해야 하며, 루프 객체를 참조하거나 메서드를 호출할 필요가 거의 없습니다. 이 절은 주로 이벤트 루프 동작을 세부적으로 제어해야 하는 저수준 코드, 라이브러리 및 프레임워크의 작성자를 대상으로 합니다.
그리고 이런 asyncio를 활용하는 함수들은 전통적인 함수와 구분하기 위해 async def라는 예약어로 함수를 선언합니다. 또한 이렇게 선언된 함수를 실행할 때에는 명시적으로 await 예약어를 붙여주어야 합니다.
asyncio 예제
다음은 asyncio 공식 문서에 있는 간단한 예제 코드입니다. say_after라는 delay와 what을 입력받으면 delay만큼 기다린 다음 what으로 입력받은 내용을 출력하는 함수를 구현합니다. 참고로 create_task 함수는 asyncio 함수를 지정된 파라미터와 함께 곧바로 실행 가능한 Task 객체로 감싸주는 함수인데요, 자세한 내용이 궁금하시다면 예제에 대한 공식 문서를 참조해주세요.
그리고 delay 파라미터에 각각 1초, 2초를 넣어 두 번 실행하였는데 두 함수의 실행이 완료되는 데 총 2초밖에 걸리지 않았습니다.
예제 코드
import asyncio
import time
async def say_after(delay, what):
await asyncio.sleep(delay)
print(what)
async def main():
task1 = asyncio.create_task(say_after(1, 'hello'))
task2 = asyncio.create_task(say_after(2, 'world'))
print(f"started at {time.strftime('%X')}")
await task1
await task2
print(f"finished at {time.strftime('%X')}")
asyncio.run(main())
Python
복사
실행 결과
started at 17:14:32
hello
world
finished at 17:14:34
Plain Text
복사
이는 같은 프로그램을 전통적인 방식으로 구현하면 총 3초가 필요한 것과 비교되는 결과입니다.
예제 코드
import time
def say_after(delay, what):
time.sleep(delay)
print(what)
def main():
print(f"started at {time.strftime('%X')}")
say_after(1, 'hello')
say_after(2, 'world')
print(f"finished at {time.strftime('%X')}")
main()
Python
복사
실행 결과
started at 17:15:32
hello
world
finished at 17:15:35
Plain Text
복사
asyncio에 대한 더 자세한 내용은 공식 문서나 다른 분들께서 잘 정리해두신 글들이 많으니 참고해보시기를 추천드립니다.
FastAPI
asyncio 기반의 여러 라이브러리를 리서치하던 중 FastAPI라는 라이브러리를 발견하고 선택하게 되었습니다. FastAPI는 Starlette이라는 ASGI 마이크로프레임워크 기반 라이브러리이며, 공식 문서에서는 가장 빠른 파이썬 웹 프레임워크라고 소개하고 있습니다.
FastAPI 라이브러리를 선택한 이유
FastAPI 라이브러리를 선택하게 된 이유는 다음과 같습니다.
1.
기존 코드베이스를 유지하는 것이 중요하였으므로, Flask와 사용 방법이 비슷하다는 것이 큰 장점이었습니다.
2.
타입 힌트를 적극적으로 활용해서 타입 검사 및 문서화를 쉽게 할 수 있는 것 역시 장점이었습니다.
•
이미 프론트엔드에서 강력한 타이핑이 가능한 타입스크립트를 사용하고 있었기 때문에 타이핑을 적극적으로 사용하는 것에 부담은 없었습니다.
3.
문서가 잘 정리되어 있고, 커뮤니티가 활성화되어있는 점도 긍정적인 요소였습니다.
4.
마지막으로 API를 구현하는데 도움이 되는 각종 편의 기능들도 편리하게 느껴졌습니다.
Flask → FastAPI 전환 가이드
지금부터는 Flask 코드를 FastAPI 코드로 전환하는 간단한 가이드를 소개드리려고 합니다. 소개해드리는 예제는 타입 힌트 같은 것들을 의도적으로 제외하여 간소하게 만든 예제이므로 이런 식으로 동작한다는 정도로만 참고해주시면 좋을 것 같습니다.
경로 라우팅
기본적으로 경로 라우팅은 Flask와 비슷하게 데코레이터 방식으로 구현하실 수 있습니다. 메서드를 지정하는 방식이 조금 다를 뿐 거의 같은 코드로 느껴질 정도로 동일합니다.
Flask
from flask import Flask
app = Flask(__name__)
@app.route('/ping/', methods=('GET',))
def ping():
return 'pong'
Python
복사
FastAPI
from fastapi import FastAPI
app = FastAPI()
@api.get('/ping/')
async def ping():
return 'pong'
Python
복사
Blueprint → APIRouter
Flask에서는 서버의 규모가 커지게 되면 서버를 여러 모듈로 나누고 하면서 여러 엔드포인트를 한데 묶어서 관리할 수 있는 Blueprint를 사용하게 됩니다. FastAPI에서는 APIRouter라는 모듈이 그 역할을 맡고 있습니다.
Flask
from flask import Blueprint
api = Blueprint(
'api', __name__, url_prefix='/api',
)
@api.route('/ping', methods=('GET',))
def ping():
return 'pong'
app.register_blueprint(api)
Python
복사
FastAPI
from fastapi import APIRouter
api = APIRouter(prefix='/api')
@api.get('/ping/')
async def ping():
return 'pong'
app.include_router(api)
Python
복사
경로 파라미터와 URL 파라미터
FastAPI는 경로 및 URL 파라미터 구현에 있어서 타입 힌트를 적극적으로 활용합니다. 덕분에 URL 파라미터의 타입을 타입 힌트로 지정해줄 수 있으며, 타입 검사에 실패하면 알아서 오류를 내 주게 됩니다.
Flask
from flask import request
@api.route(
'/people/<int:person_id>/', methods=('GET',),
)
def get_person(person_id):
try:
post_offset = int(
request.args.get('post_offset', '0'),
)
except ValueError:
raise BadRequest
...
Python
복사
FastAPI
@api.get('/people/{person_id}/')
async def get_person(
person_id: int, post_offset: int = 0,
):
...
Python
복사
글로벌 컨텍스트 → 디펜던시 인젝션
Flask에서는 설정이나 SQLAlchemy 연결 등을 글로벌 컨텍스트로 만들어서 엔드포인트에서 가져다 쓸 수 있도록 구현하곤 합니다.
FastAPI에서는 이러한 요구사항들을 디펜던시 인젝션 개념을 사용하여 해결하고, 함수 파라미터로 받아서 사용하는 자연스러운 사용법을 지원합니다. 이러한 구현 방법은 필요하지 않은 경우 디펜던시를 로드하지 않으므로 리소스 활용에 있어서 도움이 됩니다.
또한 디펜던시를 구현하면서 이미 구현된 다른 디펜던시를 재사용할 수 있는데 이를 통해 코드의 중복을 줄일 수 있습니다. 참고로 이렇게 사용할 경우 각각의 디펜던시 함수는 하나의 요청 문맥에서 한 번씩만 실행되고 이후에 사용될 때에는 그 결과를 다시 사용하게 됩니다.
Flask
from werkzeug.local import LocalProxy
@LocalProxy
def session():
return get_or_create_session()
@api.route('/people/', methods=('GET',))
def get_people():
query = session.query(Person)
return {
'people': [serialize(p) for p in query],
}
Python
복사
FastAPI
from fastapi import Depends
async def session() -> Session:
return get_or_create_session()
@api.get('/people/')
async def get_people(
session: Session = Depends(session),
):
query = session.query(Person)
return {
'people': [serialize(p) for p in query],
}
Python
복사
JSON 페이로드 직렬화와 역직렬화
Pydantic은 Python 3.7에 추가된 dataclasses 라이브러리와 비슷한 기능을 제공하면서 관련된 다양한 유틸리티 기능들을 제공합니다. 이 덕분에 API 구현 시 반복된 작업이 필요한 직렬화와 역직렬화 구현을 클래스 선언으로 대신할 수 있습니다. 뒤에서 설명하겠지만 이렇게 구현한 클래스는 API 문서 생성에도 사용됩니다.
Flask
@api.route('/people/', methods=('POST',))
def add_person():
p = request.get_json()
try:
assert isinstance(p['name'], str)
assert isinstance(p['age'], int)
except AssertionError:
raise BadRequest
person = Person(name=p['name'], age=p['age'])
add_person_to_db(person)
return serialize(person), 201
Python
복사
FastAPI
from pydantic import BaseModel
class Person(BaseModel):
name: str
age: int
@api.post('/people/', status_code=201)
async def add_person(p: Person):
add_person_to_db(person)
return person
Python
복사
엔드포인트 테스트
저희는 앞의 과정을 거쳐서 만들어진 엔드포인트를 테스트하기 위해 pytest-asyncio 라이브러리와 async-asgi-testclient 라이브러리를 사용했습니다. 두 라이브러리는 각각 pytest에서 asycnio 테스트 코드를 실행할 수 있도록 하는 기능과 asgi 기반 웹 서버를 테스트할 수 있는 기능을 제공하며, 테스트는 async-asgi-testclient 라이브러리의 TestClient 모듈을 사용하여 엔드포인트에 요청을 보내고 그 결과를 테스트하는 식으로 진행했습니다. 참고로 Starlette 라이브러리도 자체적으로 테스트를 위한 클라이언트 모듈을 제공하지만, 저희가 만든 웹소켓 엔드포인트를 테스트하기에 적합하지 않은 부분이 있어서 async-asgi-testclient 라이브러리를 사용하게 되었습니다.
여기서 디펜던시 인젝션의 장점을 하나 볼 수 있는데, FastAPI는 엔드포인트에서 사용하는 디펜던시를 테스트에서 모킹한 것으로 대신 인젝션하는 것을 지원하기 때문에, Flask의 글로벌 컨텍스트를 모킹하는 것보다 더 간단한 방법으로 SQLAlchemy 연결 등을 모킹할 수 있습니다. 자세한 내용은 다음 링크를 참조하시기 바랍니다.
Flask
@typechecked
def test_ping(fx_wsgi_app: Flask) -> None:
client = fx_wsgi_app.test_client()
resp = client.get('/ping/')
assert resp.status_code == 200
assert resp.get_data(as_text=True) == 'pong'
Python
복사
FastAPI
from async_asgi_testclient import TestClient
from pytest import mark
@mark.asyncio
async def test_ping(fx_asgi_app: FastAPI):
async with TestClient(fx_asgi_app) as client:
resp = await client.get('/ping/')
assert resp.status_code == 200
assert resp.text == 'pong'
Python
복사
FastAPI 더 잘 써 보기
FastAPI의 기능들을 조금 더 적극적으로 사용하실 수 있는 방법들을 소개해드리도록 하겠습니다.
타입 힌트의 적극적인 사용
FastAPI를 사용하겠다고 결심하셨다면, 타입 힌트를 이전보다 더 적극적으로 사용하시는 것을 추천드립니다. 아무래도 파이썬의 타입 힌트는 런타임을 위해 도입된 기능은 아니다보니 별도로 런타임 타입 체크 라이브러리 같은 걸 쓰지 않는 이상 적으면 좋고 아니면 말고의 느낌이 강했는데요. FastAPI에서는 타입 힌트를 적으면 자동화된 문서화와 타입 검증이 가능해지기 때문에 적극적으로 작성하시기를 추천드립니다. 앞서 말씀드린 것과 같이 경로 파라미터와 쿼리 파라미터, 그리고 페이로드에 대한 타입 명시를 타입 힌트를 사용해 하실 수 있습니다.
예제
아래와 같이 타입 힌트와 데코레이터에 타입을 적극적으로 명시하면 우측과 같이 매우 쓸만한 API 문서가 만들어집니다. 저는 이 덕분에 별도로 API 문서를 만들 필요가 없어졌으며 협업을 할 때에도 문서를 보여드리면 되어서 편리했습니다. 만들어지는 Swagger 문서의 경우 문서에서 바로 API를 직접 실행해볼 수도 있어서 개발 편의성 역시 크게 증대되었습니다.
@api.post(
'/people/',
response_model=CreatePersonResponse,
status_code=status.HTTP_201_CREATED,
responses={
400: {
'description': 'invalid name',
'content': {'application/json': {...: ...}},
}
},
)
async def create_person(
payload: Person,
session: Session = Depends(session),
):
...
Python
복사
디펜던시 인젝션의 활용
FastAPI의 디펜던시 인젝션은 SQLAlchemy 연결 같은 것뿐만 아니라 공통된 파라미터 검증 로직을 구현하는 데에도 사용할 수 있습니다.
아래의 예시는 헤더의 토큰을 검증하는 로직을 디펜던시로 만들어 APIRouter의 디펜던시로 선언하는 코드입니다. 이렇게 하면 토큰 검증이 필요한 엔드포인트에서 매번 토큰을 검증하는 코드를 작성할 필요가 없어집니다. '토큰을 통해 DB에서 유저 정보를 가져오는 것'과 같이 자주 사용될 것 같은 로직은 기존 디펜던시를 의존하는 새로운 디펜던시를 만들어서 구현하는 것도 좋은 방법입니다.
from fastapi import Header
async def verify_token(x_token: str = Header(...)):
if x_token != 'fake-super-secret-token':
raise HTTPException(status_code=401)
return x_token
async def verify_token_get_user(
token: str = Depends(verify_token),
session: Session = Depends(session),
):
return get_user_from_token(token, session)
auth_api = APIRouter(
dependencies=[Depends(verify_token)],
)
@auth_api.get('/authorized-ping/')
async def authorized_ping():
return 'pong'
@auth_api.get('/me/')
async def me(
u: User = Depends(verify_token_get_user),
):
return {'name': u.name}
Python
복사
웹소켓 엔드포인트 구현
FastAPI를 도입하게 된 근본적인 이유인 웹소켓 엔드포인트 구현은 FastAPI에서 자체적으로 제공하는 웹소켓 모듈을 통해 쉽게 구현할 수 있었습니다. 문서를 보고 웹소켓 로직을 구현하기만 하면 asyncio의 도움으로 통신 대기 시간에도 리소스를 필요한 곳에 사용하는 효율적인 웹소켓 엔드포인트를 만들 수 있습니다.
from fastapi import WebSocket
@api.websocket('/ws/')
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f'echo: {data}')
Python
복사
asyncio 더 잘 써보기
FastAPI를 도입하고 asyncio를 활용한 엔드포인트로 전환을 하고 보니 기존에 사용하고 있던 I/O 라이브러리들이 눈에 들어오기 시작했습니다. 아무리 asyncio 엔드포인트를 만들어 두었어도 내부에서 asyncio를 활용하지 않는 I/O 작업을 실행하면 대기 시간을 활용하지 못하기 때문입니다. 그래서 asyncio를 활용할 수 있는 라이브러리들을 찾아보기 시작하였습니다.
asyncpg — PostgreSQL 드라이버
먼저 소개해드릴 라이브러리는 asyncpg 라이브러리입니다. SQLAlchemy 버전 1.4 이상에서 사용하면 데이터베이스 요청 작업을 asyncio 이벤트 루프 위에서 돌릴 수 있게 됩니다.
아시다시피 보통의 웹 서비스의 백엔드는 데이터베이스 I/O가 가장 많이 일어나는 I/O라고 해도 과언이 아닐 것입니다. SQLAlchemy는 1.4부터 asyncio를 지원하는 드라이버를 지원하기 시작했습니다. 다만 문서상으로는 아직 베타 레벨이며 하위 호환성을 깨뜨리는 업데이트가 있을 수 있으니 유의하라는 내용이 있으니 참고 바랍니다.
The asyncio extension as of SQLAlchemy 1.4.3 can now be considered to be beta level software. API details are subject to change however at this point it is unlikely for there to be significant backwards-incompatible changes.
그래서 현재 PostgreSQL을 사용 중이시라면 asyncpg를 도입하여 asnycio를 활용한 데이터베이스 요청 작업을 수행하실 수 있습니다. 대신 명시적으로 await 예약어를 붙여야만 쿼리와 같은 I/O 작업을 수행할 수 있게 되므로 ORM을 사용할 수 있는 방식이 크게 제한되게 됩니다. 대표적으로 2.0 스타일이라고 언급되는 쿼리 스타일을 사용하는 것이 사실상 강제되며, 관계된 인스턴스를 사용하려면 selectinload와 같은 eager loading 옵션을 필수적으로 사용해야 합니다. 그럼에도 asyncio를 사용하면서 성능상 이점이 있고 레거시 로직을 실행할 수 있는 run_sync 같은 함수도 제공하고 있으니 도입을 고려해보셔도 좋을 듯합니다.
그밖의 라이브러리
그 외에 저희가 사용 중인 asyncio 라이브러리들을 아래에 간단히 소개드립니다.
aioredis — Redis 클라이언트 라이브러리
aiopika — RabbitMQ 클라이언트 라이브러리
결론
저희처럼 Flask를 백엔드에 사용하고 계시면서 I/O 성능 개선의 필요성을 느끼셨다면 FastAPI를 도입하는 것이 좋은 대안이 될 수 있을 것입니다. 꼭 성능 때문이 아니더라도, API를 구현하시면서 부수적인 코드를 반복적으로 구현하시는 것에 지치셨다면 그러한 문제를 해결해줄 수 있는 이름에 걸맞은 대안이기도 합니다. 사용법도 많이 다르지 않으니 가볍게 시도해보세요.
그리고 asyncio를 지금까지 사용하지 않으셨다면, 이제는 도입해봐도 좋은 시기가 되지 않았나 생각합니다. FastAPI로 전환하는 것이 부담되신다면 Flask의 asyncio 엔드포인트 지원 사용을 검토해보세요. 개인적으로는 제가 알아보기 시작했을 때 Flask의 asyncio 지원이 있었다면 빠른 릴리스를 위해 Flask를 계속 사용했을 것 같기도 합니다. SQLAlchemy 역시 추후 2.0 버전이 나오면 2.0 스타일의 쿼리만 사용할 수 있게 될 예정인데, 그렇게 되면 앞서 언급한 제한들이 많이 사라지게 되므로 asyncio 기반의 드라이버를 사용하는 것이 더 보편적인 선택이 될 것으로 예상됩니다.
마지막으로 저도 이번에 작업을 진행하면서 알게 되었지만, 커뮤니티의 활발한 기여로 이미 많은 asyncio 기반 라이브러리들이 나와 있다는 것을 알 수 있었습니다. 다만 asyncio와 같은 비동기 라이브러리를 사용해보신 적이 없으시다면 오류 발생 시 대처가 힘드시거나 사용에 어려움이 있으실 수 있으므로 공식 문서를 정독해보시기를 권장드립니다.
제가 준비한 내용은 여기까지이며, 다음에는 더 좋은 글로 인사드리도록 하겠습니다.
읽어주셔서 감사합니다.