1. 정리의 목적

FastAPI? 요새 다 좋다고 소문났던데? 빠르고 가벼운 프레임워크? 비동기?

라는 추상적인 내용에 가볍게 프로젝트를 시작해보는 경우가 많습니다.

공식 문서를 참고하며 프로젝트를 시작하기 전에 보면 도움될 만 한 내용을 정리하였습니다.


2. FastAPI란

고성능, 간편한 학습, 빠른 코드 작성, Python3.6+의 API를 빌드하기 위한 준비된 프로덕션 웹프레임워크

라고 공식 문서에 나와있습니다.


  • 빠른 속도: Starlette와 Pydantic 덕분에 Go와 NodeJS와 대등할 정도의 높은 성능
  • 빠른 개발 속도: 기존에 비해 2~3배 더 빠른 개발 속도
  • 적은 버그: 휴먼 에러가 40% 감소
  • 배우기 쉬움: 쉽게 사용하고 친절한 문서
  • 견고함: 대화형 문서 지원 (OpenAPI 기반으로 JSON 스키마와 완벽한 호환)

2-1. 다양한 스폰서 및 사용 회사

2-2. 맛보기 (FastAPI & Flask)

Flask의 엔드포인트를 한데 묶어 관리하는 Blueprint의 역할을 FastAPI는 APIRouter가 담당


Flask

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from flask import Blueprint

api = Blueprint(
    'api', __name__, url_prefix='/api',
)

@api.route('/check', methods=('GET',))
def get_check():
    return 'check'

app.register_blueprint(api)

FastAPI

1
2
3
4
5
6
7
8
9

from fastapi import APIRouter

api = APIRouter(prefix='/api')

@api.get('/check/')
async def get_check():
    return 'check'

app.include_router(api)

3. 주요한 기능 3가지

  • 높은 속도와 성능 (Starlette과 Pydantic)
  • ASGI를 기반한 비동기 (async/await)
  • 코드 변화에 따른 요청/응답 스키마 문서화(openAPI)

3-1. 높은 속도와 성능

3-1-1. Starlette

Python 3.6 이상의 비동기적 웹 애플리케이션 개발을 위한 경량화된 프레임워크(ASGI 프레임워크 툴킷)


FastAPI는 왜 Starlette 무슨 연관이 있을까?

  • FastAPI는 내부적으로 Starlette를 사용
    • 내부적으로 감싸서 활용
      • Starlette를 직접 사용하는 것에 비하면 성능적으로 부족하지만 타 프레임워크보다 월등한 성능
    • Starlette, FastAPI 모두 Uvicorn을 사용

Uvicorn

  • 파이썬 전용 ASGI (Asynchronous Server Gateway Interface)
  • uvloop와 httptools를 사용하는 비동기 웹서버 (속도의 비밀 → libuv, Cython)
    • 설치 명령 $ pip install "uvicorn[standard]" -> 추천
      • Cython 기반의 디펜던시 설치
      • 이벤트 루프인 uvloop가 사용되고 http 프로토콜은 httptools로 처리
    • 설치 명령 $ pip install uvicorn -> 비추
      • uvloop가 설치되지 않고 event loop로 asyncio 사용 → 성능 저하
        • uvloop -> C 기반, asyncio -> 파이썬 기반
  • WSGI 애플리케이션은 요청 → 응답을 반환하는 단일 동기 호출
    • 긴 폴링 HTTP 요청, WebSocket등 연결 지속성을 허용하지 않는 단점
      • 비동기 동시성 ASGI를 사용하면 가벼운 백그라운드 작업과 네트워크 I/O에서 많은 시간 소요되는 엔드포인트에 대한 리스크를 줄여줌
  • HTTP/1.1 및 WebSocket 지원

[ASGI Framework]FastAPI → Starlette → [ASGI Server]uvicorn → uvloop(cython) → libuv

  1. FastAPI는 Starlette를 사용하기 때문이고, Starlette가 빠른 이유는
  2. Uvicorn을 사용하기 때문이고, Uvicorn이 빠른 이유는
  3. Uvloop를 쓰기 때문이고, Uvloop가 빠른 이유는
  4. Node.js 비동기 I/O 핵심인 libuv를 기반으로 Cython으로 작성되었기 때문


3-1-2. Pydantic

  • FastAPI의 입출력 스펙을 정의하고 값을 유효성 검사를 위해 사용하는 라이브러리
  • 스펙을 검증할때는 갯수, 타입, 필수값 여부등 설정 가능하며 json, yaml등과 같이 직렬화 형식 지원
  • Pydantic의 라이브러리는 Cython으로 컴파일되어 최대 50% 이상 성능 향상
  • Request, Response Model 정의
  • 모델 클래스 생성 후 상속을 통해 코드 중복 해결

Request, Response Model 예제

pydantic의 클래스내 멤버 변수들은 default로 required이며 optional로 설정 가능

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from pydantic import BaseModel, Field, Query, Optional


# request query params
class AreaInfoParams(BaseModel):
    parent_area_: Optional[int] = Query(None, title='상위 구역')
    child_area: Optional[int] = Query(None, title='하위 구역')

# request query body
class CreateAreaBody(BaseModel):
    name: str = Field(title='이름')
    area_id: str = Field(title='구역 ID')
    description: str = Field(title='구역 설명')
    limit_count: int = Field(title='구역 제한 수량', default=10)

# response object model
class AreaDetailItem(BaseModel): # BaseModel 상속
    name: str | None = Field(title='이름')
    description: str = Field(title='구역 설명')

클래스 모델 선언 후 다른 속성값으로 사용 가능

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from pydantic import BaseModel, Field


class Category(BaseModel):
    name: str | None = Field(title='이름')
    

class Post(BaseModel):
    category: Category = Field(title='카테고리') # 카테고리 클래스 선언 
    title: str = Field(title='제목')
    content: str = Field(title='내용')

@validator

field의 매개변수로 gt, lt, min, max등의 기본 유효성 검사 대신 사용자 정의 유효성 검증 데코레이터

1
2
3
4
5

@validator('length') # 여러개 문자열로 선언 가능
def validate_length(cls, value):
    if value > LIMIT_LENGTH:
        return LIMIT_LENGTH
    return value

@root_validator

전체 모델의 유효성 검증을 위한 데코레이터

(Ex: 리스트를 전달 받았을시 전체 항목에 중복 제거 및 상태값 필터 적용시)

1
2
3
4
5

@root_validator
def check_validation_fields(cls, values: dict[int, Any]) -> dict[int, Any]:
    if all(map(lambda x: x > 0, values.values())):
        raise InvalidParamException()
    return values

3-1-3. DI

  • FastAPI는 DI(Dependency Injection)을 지원
  • 의존성 주입을 사용해 애플리케이션내 객체 인스턴스화 후 재사용 및 관리 가능
  • 의존성 주입을 위해 FastAPI는 Depends 데코레이터를 제공

Depends

  • FastAPI에서 의존성 주입을 지원하는 함수
  • 동작 순서
    1. 함수를 실행하기 전 먼저 실행되며 Depends 내부에 함수를 매개변수로 전달
    2. 전달된 함수의 파라미터를 주입하여 로직 구현
    3. 함수 리턴 -> 재사용 및 유지보수성 향상
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from fastapi import FastAPI, Depends

app = FastAPI()


def query_validate(q: str = None) -> str:
    if q is None or q == '':
        return 'hello'
    return q


def get_query_parameter(q: str = None) -> str:
    return query_validate(q)


@app.get("/search/")
async def search(query_param: str = Depends(get_query_parameter)):
    return {"query_param": query_param}


3-2. 비동기 async/await

사용 가능 조건

  • 블로킹되는 I/O 바운드 작업
    • 외부 자원에 의존하여 작업이 완료될때까지를 기다리는 케이스
      • 네트워크 호출, 파일 읽기쓰기, DB 쿼리 조회등
        • Ex) 파일 읽는 동안 다른 작업을 수행하지 못하는 경우 → 블로킹
  • async/await를 지원하는 라이브러리를 사용하는 경우
    • async def를 이용해 작성된 함수는 코루틴에 등록되는데 FastAPI는 코루틴을 사용해 비동기 최적화
    • await 사용을 지원하지 않는 서드파티 라이브러리 사용의 경우 기본 def 사용
  • db에서의 비동기 sqlalchemy 설정
    • SQLAlchemy 1.4버전부터 비동기 처리 지원
      • create_async_engine, aiomysql 설정
      • session 생성시 AsyncSession 주입
      • asyncio 모듈을 사용하여 비동기 DB 작업
        • async with(컨텍스트 관리자) 구문 사용해 처리, 이때 자원을 소비하지 않은 채로 CPU 자원을 반납 (이후 다른 코드 실행 가능)

3-2-1. 블로킹되는 I/O 바운드 작업

  • $ uvicorn main:app
  • $ uvicorn main:app --workers 3
    • $ ps -ef | grep {pid} (워커의 갯수 확인 명령)
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 블로킹 I/O 예제
# async를 사용하여 내부 스레드풀에서 관리되기에 블로킹된다.

@app.get('/')
async def root():  # 내부 스레드 풀: fastAPI 내부적 관리(비동기 처리)
    """
    내부 스레드풀 사용시 실행하는데 별도의 스레드풀을 생성하지 않고 FastAPI 내부적으로 최적화 관리
    """
    print(os.getpid())
    while True:
        pass

# async를 사용하지 않은 즉, 외부 스레드풀을 사용하여 서버가 블로킹 되지 않는다.
@app.get('/')
def root():  # 외부 스레드 풀: 외부 스레드는 운영체제, 라이브러리등에서 생성/관리
    """
    외부 스레드 풀에서 다이렉트 실행
    CPU, 복잡한 연산작업등이 해당되며 fastapi에서 사용자가 직접 생성하는 스레드풀
    많은 제어력을 제공하며 스레드풀의 크기, 생성 및 소멸등의 제어 가능
    """
    print(os.getpid())
    while True:
        pass

3-2-2. 비동기 함수 동작 순서

  1. 이벤트 루프에서 함수 호출
  2. 이벤트 루프는 해당 함수가 완료될때까지 다른 작업 수행 가능 (I/O 작업과 같은 블로킹 작업 수행시 유용)
  3. 함수값이 리턴될 준비가 되면 이벤트 루프가 해당 함수의 작업을 재개

3-2-3. async&await 샘플 코드

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, Session
from database import get_db

app = FastAPI()

# 비동기 SQLAlchemy 엔진 객체 생성
async_engine = create_async_engine(get_db())

# 비동기 세션 클래스 AsyncSession 생성
async_session = sessionmaker(
    async_engine,
    expire_on_commit=False,
    class_=AsyncSession
)


async def get_db_session() -> AsyncSession:
    async with async_session() as session:
        yield session


@app.post("/users/")
async def create_user_route(user: UserCreate, db: AsyncSession = Depends(get_db_session)):
    db_user = await create_user(db=db, user=user)
    return db_user

@app.get("/users/{user_id}")
async def read_user(user_id: int, db: AsyncSession = Depends(get_db_session)):
    db_user = await get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

3-3. 코드 변화에 따른 요청/응답 스키마 문서화

FastAPI는 API를 구현하기만 하면 자동으로 요청/응답 스키마를 문서화로 제공합니다.

이를 통해 문서 작업에 할애하는 시간을 단축시키고 기능 구현에만 집중할 수 있도록 지원합니다.


3-3-1. url

  • /docs
    • swagger UI와 유사 기능 제공
    • API path, 매개변수, 스키마등의 정보를 문서로 자동 생성 (api 테스트 가능)
  • /redoc
    • openAPI spec 파일을 읽어 문서를 반영해주는 도구
    • swagger UI와 유사 기능 제공
  • /openapi.json

3가지 문서를 지원하고 OpenAPI를 채택하여 이를 기반해 규격에 맞는 json 파일을 생성합니다.



용어 및 개념

WSGI

Web Server Gateway Interface란 뜻으로 웹서버와 웹프레임워크 사이에 통신을 담당한다.

웹서버는 html, js외에 서버쪽 python, java등으로 구성된 프레임워크 언어를 해석할 수 없다.

이를 해결해주는 것이 wsgi의 역할이고 wsgi는 비동기 방식의 콜백함수로 이루어져

쓰레드를 생성하지 않고도 여러 요청을 동시에 처리할 수 있다.

uwsgi, gunicorn이 대표적이다.

  • 웹서버 요청과 앱의 응답을 번역 및 http 응답으로 변환
  • 요청이 들어오면 파이썬 코드 호출

ASGI

WSGI와 비슷한 구조를 가지나 단점을 보완할 수 있게 모든 요청을 비동기로 처리한다.

WSGI에선 지원하지 않는 HTTP/2.0을 지원한다.

오랜 시간 연결을 유지하는 websocket이나 긴 HTTP 요청을 처리하는데 유리하다.


Cython

파이썬과 C언어를 혼합한 하이브리드 언어이다.

파이썬 코드를 C언어로 변환하여 속도를 향상 시키는 기술을 의미한다.

쉽게 말해 C언어 코드로 변환하면서 C언어의 성능 최적화 이점을 얻는 효과를 발휘한다.


Uvloop

uvloop는 asyncio를 대체하기 위해 만들어졌다.

Cython으로 작성되었으며 libuv 위에서 구현되었다. (libuv는 nodejs의 고성능 플랫폼 비동기 I/O 라이브러리)

low level의 언어로 구현된 것들을 파이썬 객체를 통해 래핑하여 구현되었다. (관련문서)


Coroutine

코루틴은 다양한 진입점, 탈출점이 있는 루틴 (Multiple 탈출점)

1
2
3
4
5

# 서브 루틴: 호출된 함수 자체 
def get_sum(a: int, b: int) -> int:
    return a + b
  
print(get_sum(1, 3)) # 루틴: 함수 호출부

루틴은 서브루틴이 리턴할때까지 대기를 해야하기 때문에 루틴과 서브루틴은 서로 종속적 관계에 있다고 볼 수 있다. (서브루틴에서 리턴 후 루틴이 값을 활용 가능)

파이썬에서 코루틴은 루틴에 서브루틴에 종속적이지 않고 대등한 관계로 순차적 호출이 가능한 함수이다.


네트워크, 디스크 I/O, db 쿼리, 원격 api 처리를 위해 대기시 다른 작업을 먼저 처리함으로써 유후 시간을 최소화하기 때문에 스레드내 효율성을 증가시킬 수 있다.

  • 코루틴 함수: async def 으로 정의한 함수
  • 코루틴 객체: 코루틴 함수를 호출하고 반환되는 객체

파이썬에서 코루틴은 대표적으로 함수 앞에 붙는 async 키워드 이며, asyncio 를 통해 사용할 수 있다. (관련 영상)


Event loop

파이썬 내부 스레드는 이벤트 루프에 의해 제어되고

이벤트 루프는 스레드간 작업 스케줄링과 이벤트 처리를 담당하는 기능을 한다.

(Ex: 비동기 태스크 및 콜백을 실행하고 네트워크 IO 연산을 수행하며 자식 프로세스를 실행)


실행 순서

  1. 이벤트 루프는 task들을 loop를 돌면서 하나씩 실행
  2. 실행한 task가 특정 데이터 요청 후 응답 대기 상태라면 task는 제어권을 다시 이벤트 루프에 넘겨준다.
  3. 제어권을 받은 이벤트 루프는 다음 테스크를 실행하게 되고 응답을 받은 순으로 테스크 큐에 적재된다.
  4. 재개되는 테스크들은 멈췄던 제어권을 가지고 작업을 마무리한다.

여기서 코루틴이 응답을 대기하는 상태에서 제어권을 이벤트 루프로 주는 용도로 await를 사용한다.


DI

사용하는 객체가 아닌 외부의 독립적 객체로 인스턴스 생성 후 전달해 의존성을 해결하는 방법을 말한다.

쉽게 말해 상위 계층과 하위 계층에 대한 영향도를 줄이는 개념이며,

의존성 주입을 하는 이유는 결합도를 낮춰서 변경에 따른 비용을 축소시키고

다른 객체와의 관계에만 더 집중할 수 있도록 해주기 때문이다. (관심사 분리)



References