FastAPI로 빠르고 강력한 API 구축하기: 비동기 처리와 자동 Swagger UI 활용

 

FastAPI는 Python으로 RESTful API를 빠르고 쉽게 구현할 수 있는 현대적인 프레임워크로, 비동기 지원과 자동 Swagger UI 같은 강력한 기능을 제공합니다. 아래에서 이를 구체적으로 설명하고, 독립적인 예제를 통해 사용 방법을 보여드리겠습니다.

---

FastAPI란?

FastAPI는 Python 3.6 이상에서 동작하며, Starlette(비동기 웹 프레임워크)과 Pydantic(데이터 검증 및 직렬화)을 기반으로 설계되었습니다. Flask와 달리 비동기 처리를 기본으로 지원하며, API 문서 자동 생성과 타입 힌트를 활용한 코드 가독성이 큰 장점입니다.

---

주요 특징: 비동기 지원과 자동 Swagger UI

1. 비동기 지원

FastAPI는 Python의 asyncio와 async/await 구문을 활용해 비동기 처리를 지원합니다. 이를 통해 I/O-bound 작업(예: 데이터베이스 쿼리, 외부 API 호출)에서 높은 성능을 발휘합니다.

• 왜 유용한가?: 동기 방식(Flask 등)에서는 요청이 순차적으로 처리되지만, 비동기 방식은 여러 요청을 동시에 처리해 처리량을 늘릴 수 있습니다.
• 예시: 네트워크 요청이나 파일 I/O가 많은 경우, FastAPI는 병렬 처리를 통해 응답 시간을 단축합니다.

2. 자동 Swagger UI

FastAPI는 API 엔드포인트를 정의하면 자동으로 Swagger UI(인터랙티브 API 문서)를 생성합니다. 별도의 설정 없이 /docs 경로에서 확인 가능하며, 요청/응답 스키마와 테스트 UI를 제공합니다.

• 왜 유용한가?: 개발자나 클라이언트가 API를 쉽게 이해하고 테스트할 수 있어 문서 작성 시간을 줄이고 협업 효율성을 높입니다.
• 대안 비교: Flask는 Swagger를 위해 Flask-RESTPlus 같은 추가 라이브러리가 필요하지만, FastAPI는 기본 내장.

---

 

FastAPI 설치 및 기본 예제

1. 설치

bash

 

pip install fastapi uvicorn

• fastapi: 프레임워크 자체.
• uvicorn: ASGI 서버로, FastAPI를 실행하는 데 필요.

2. 기본 코드 (main.py)

python

 

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

# 데이터 모델 정의
class Item(BaseModel):
    id: int
    name: str

# 메모리 내 데이터
items = [
    Item(id=1, name="Item 1"),
    Item(id=2, name="Item 2")
]

# GET: 모든 아이템 조회 (비동기)
@app.get("/items", response_model=List[Item])
async def get_items():
    return items

# GET: 특정 아이템 조회 (비동기)
@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    item = next((item for item in items if item.id == item_id), None)
    if item:
        return item
    return {"error": "Item not found"}

# POST: 아이템 추가 (비동기)
@app.post("/items", response_model=Item, status_code=201)
async def create_item(item: Item):
    items.append(item)
    return item

3. 실행

bash

 

uvicorn main:app --reload

• 서버가 http://localhost:8000에서 시작됩니다.

---

비동기 지원 확인

위 코드에서 @app.get()과 @app.post()는 모두 async def로 정의되었습니다. 예를 들어, 외부 API 호출이나 DB 쿼리를 비동기로 처리할 수 있습니다:

비동기 예제 (DB 모방)

python

 

import asyncio

@app.get("/items/async-demo")
async def async_demo():
    await asyncio.sleep(2)  # 비동기 작업 시뮬레이션 (예: DB 쿼리)
    return {"message": "This took 2 seconds but didn't block other requests!"}

• 여러 클라이언트가 동시에 요청해도 서로를 기다리지 않고 병렬 처리.

---

자동 Swagger UI 확인

1. 서버 실행 후 브라우저에서 http://localhost:8000/docs 접속.
2. Swagger UI가 표시되며, 정의된 엔드포인트(/items, /items/{item_id}, /items/async-demo)를 확인 가능.
3. 각 엔드포인트 옆의 "Try it out" 버튼으로 직접 요청 테스트 가능.

Swagger UI의 장점

• 요청 파라미터와 응답 스키마 자동 표시.
• Pydantic 모델 기반으로 데이터 유효성 검사 결과 반영.
• 별도 문서 작성 없이 바로 사용 가능.

---

 

728x90

CORS 처리 (보너스)

FastAPI도 기본적으로 CORS를 지원하지 않으므로, Vue.js와 연동하려면 설정이 필요합니다.

설치 및 설정

bash

 

pip install fastapi[all]  # CORS 포함

python

 

from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# CORS 설정
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Vue.js 출처
    allow_credentials=True,
    allow_methods=["*"],  # 모든 HTTP 메서드 허용
    allow_headers=["*"],  # 모든 헤더 허용
)

---

FastAPI vs Flask

기능

FastAPI

Flask

비동기 지원	기본 지원 ( async/await )	미지원 (동기 방식)
Swagger UI	자동 생성	추가 설정 필요
타입 힌트	강력 지원 (Pydantic)	없음
성능	비동기로 인해 우수	동기 방식으로 제한적
학습 곡선	약간 높음	매우 낮음

---

결론

FastAPI는 비동기 지원과 자동 Swagger UI가 필요한 경우 강력 추천됩니다. 비동기는 높은 트래픽 환경에서 성능을 최적화하고, Swagger UI는 API 문서화와 테스트를 간소화합니다. Flask보다 설정이 약간 복잡할 수 있지만, 생산성과 확장성을 고려하면 투자할 가치가 충분합니다. 추가로 궁금한 점 있으면 말씀해주세요!

 

---

FastAPI에서 비동기 처리를 활용해 Oracle DB의 저장 프로시저 소스를 가져오는 예제를 작성하겠습니다. 이전에 Spring Boot로 구현했던 "저장 프로시저 소스 조회" 기능을 FastAPI로 옮기고, asyncio를 사용해 비동기적으로 DB 쿼리를 처리하도록 수정합니다.

---

전제 조건

• Oracle DB에 접근하려면 cx_Oracle을 사용하며, 비동기 처리를 위해 databases 라이브러리와 함께 설정합니다.
• 하지만 cx_Oracle 자체는 비동기 지원이 제한적이므로, 여기서는 비동기 모방을 위해 asyncio.sleep()을 유지하며 실제 DB 연결은 동기적으로 처리합니다. (완전 비동기를 위해선 aiomysql 같은 비동기 DB 드라이버가 적합하지만, Oracle의 경우는 제한적입니다.)

---

FastAPI 코드: 저장 프로시저 소스 가져오기

1. 설치

bash

 

pip install fastapi uvicorn cx_Oracle databases

2. 코드 (main.py)

python

 

from fastapi import FastAPI
import cx_Oracle
import asyncio

app = FastAPI()

# Oracle DB 연결 설정 (동기적으로 초기화)
dsn = cx_Oracle.makedsn("localhost", 1521, service_name="ORCL")
connection = cx_Oracle.connect(user="your_username", password="your_password", dsn=dsn)

@app.get("/procedure-source/{procedure_name}")
async def get_procedure_source(procedure_name: str):
    # 비동기 작업 시뮬레이션 (예: 네트워크 지연)
    await asyncio.sleep(1)  # DB 쿼리 대기 시간 모방
    
    # Oracle DB에서 저장 프로시저 소스 조회 (동기적으로 실행)
    try:
        cursor = connection.cursor()
        query = "SELECT TEXT FROM USER_SOURCE WHERE NAME = UPPER(:name) AND TYPE = 'PROCEDURE' ORDER BY LINE"
        cursor.execute(query, {"name": procedure_name.upper()})
        
        # 결과 수집
        source_lines = [row[0] for row in cursor.fetchall()]
        cursor.close()
        
        if source_lines:
            return {"source": "\n".join(source_lines)}
        else:
            return {"error": "Procedure not found"}, 404
    except cx_Oracle.Error as e:
        return {"error": f"Database error: {str(e)}"}, 500

@app.on_event("shutdown")
async def shutdown():
    # 서버 종료 시 DB 연결 닫기
    connection.close()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

---

3. 실행

bash

 

python main.py

• 서버가 http://localhost:8000에서 시작됩니다.

---

4. 테스트

요청

bash

 

curl http://localhost:8000/procedure-source/SAMPLE_PROC

예상 응답 (성공 시)

json

 

{
    "source": "CREATE OR REPLACE PROCEDURE SAMPLE_PROC AS\nBEGIN\n    DBMS_OUTPUT.PUT_LINE('Hello, this is a sample procedure!');\nEND SAMPLE_PROC;"
}

예상 응답 (실패 시)

json

 

{
    "error": "Procedure not found"
}

---

코드 설명

1. 비동기 모방:
   • await asyncio.sleep(1): DB 쿼리 실행 전 네트워크 지연을 모방합니다. 여러 요청이 동시에 들어와도 다른 요청을 블록하지 않습니다.
   • 실제 비동기 DB 호출은 cx_Oracle의 한계로 동기적으로 처리되지만, FastAPI의 비동기 프레임워크 덕분에 다른 작업(예: 추가 요청 처리)은 병렬로 진행 가능.
2. 저장 프로시저 소스 조회:
   • USER_SOURCE 뷰에서 NAME과 TYPE='PROCEDURE' 조건으로 소스를 조회.
   • 결과를 줄 단위로 합쳐 반환.
3. DB 연결 관리:
   • 서버 시작 시 연결을 생성하고, 종료 시 @app.on_event("shutdown")으로 안전하게 닫음.

---

한계 및 개선 방안

• cx_Oracle의 동기적 특성: Oracle DB의 비동기 지원이 부족해 쿼리 자체는 동기적으로 실행됩니다. 완전 비동기를 원한다면:
  • aiopg (PostgreSQL)나 aiomysql (MySQL) 같은 비동기 드라이버가 있는 DB로 전환.
  • 또는 asyncio의 run_in_executor로 동기 코드를 비동기적으로 래핑:
    python

     

    from functools import partial
    
    @app.get("/procedure-source/{procedure_name}")
    async def get_procedure_source(procedure_name: str):
        loop = asyncio.get_running_loop()
        source = await loop.run_in_executor(None, partial(fetch_procedure_source, procedure_name))
        return source
    
    def fetch_procedure_source(procedure_name):
        cursor = connection.cursor()
        query = "SELECT TEXT FROM USER_SOURCE WHERE NAME = UPPER(:name) AND TYPE = 'PROCEDURE' ORDER BY LINE"
        cursor.execute(query, {"name": procedure_name.upper()})
        source_lines = [row[0] for row in cursor.fetchall()]
        cursor.close()
        return {"source": "\n".join(source_lines)} if source_lines else {"error": "Procedure not found"}

• CORS 추가: Vue.js와 연동하려면 CORS 설정 필요:
  python

   

  from fastapi.middleware.cors import CORSMiddleware
  
  app.add_middleware(
      CORSMiddleware,
      allow_origins=["http://localhost:3000"],
      allow_methods=["*"],
      allow_headers=["*"],
  )

---

결론

FastAPI로 저장 프로시저 소스 조회를 비동기적으로 구현한 예제를 제공했습니다. asyncio.sleep()으로 비동기성을 강조했으며, Oracle DB 접근은 동기적이지만 FastAPI의 비동기 프레임워크를 활용해 다른 요청을 막지 않습니다. 추가 개선이나 질문 있으면 말씀해주세요!