FastAPI 6일차: 파이썬 비동기 완전 정복 (Async/Await, 비동기 ORM, 의존성 주입)

안녕하세요! FastAPI 학습 여정 6일차입니다. 오늘은 FastAPI의 가장 강력한 기능 중 하나인 비동기(Asynchronous) 프로그래밍에 대해 깊이 파고들어 보겠습니다. 웹 애플리케이션의 성능을 한 단계 끌어올릴 수 있는 핵심 개념부터, 실제 데이터베이스 연동을 비동기로 전환하는 방법, 그리고 코드의 구조를 아름답게 만들어주는 의존성 주입까지, 오늘 다룰 내용는 여러분을 한 단계 더 높은 레벨의 개발자로 만들어 줄 것입니다.

---

1. 왜 비동기 프로그래밍이 중요한가?

현대의 웹 애플리케이션은 수많은 동시 요청을 처리해야 합니다. 만약 하나의 요청이 데이터베이스 조회나 외부 API 호출처럼 오래 걸리는 작업(I/O-bound)을 처리하는 동안 다른 모든 요청이 기다려야 한다면, 사용자 경험은 끔찍할 것입니다.

비동기 프로그래밍은 바로 이 문제를 해결합니다. 어떤 작업이 대기하는 동안, 다른 작업을 처리하여 시스템 자원을 효율적으로 사용하고 전체적인 처리량을 극대화하는 방식입니다. FastAPI는 이러한 비동기 처리를 핵심 철학으로 삼고 있으며, Python의 asyncio 라이브러리를 기반으로 합니다.

asyncio의 3가지 핵심 요소: 코루틴, await, 이벤트 루프

1. 코루틴 (Coroutines): async def로 정의된 특별한 함수입니다. 일반 함수와 달리, 중간에 실행을 멈추고 다른 코루틴에게 제어권을 넘겨줄 수 있습니다.
2. await: 코루틴의 실행을 잠시 멈추고, await 뒤에 오는 작업이 완료될 때까지 기다리게 만드는 키워드입니다. 중요한 것은, 기다리는 동안 다른 코루틴이 실행될 수 있다는 점입니다.
3. 이벤트 루프 (Event Loop): 비동기 작업의 "지휘자"입니다. 여러 코루틴의 실행 순서를 관리하고, 어떤 코루틴을 실행하고 어떤 코루틴을 대기시킬지 스케줄링하는 역할을 합니다.
   

이 세 가지 요소가 어우러져, 마치 여러 작업이 동시에 처리되는 것처럼 보이는 마법이 일어납니다.

import asyncio

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    print("--- 프로그램 시작 ---")
    await asyncio.gather(
        say_after(1, "안녕하세요"),
        say_after(2, "FastAPI!")
    )
    print("--- 프로그램 종료 ---")

asyncio.run(main())

# 출력:
# --- 프로그램 시작 ---
# 안녕하세요 (1초 후)
# FastAPI! (2초 후)
# --- 프로그램 종료 ---

위 코드에서 asyncio.gather는 두 개의 say_after 코루틴을 동시에 실행합니다. 전체 프로그램은 2초 만에 종료되는데, 이는 asyncio.sleep()이 실행되는 동안 이벤트 루프가 다른 작업을 처리하기 때문입니다.

---

2. FastAPI와 비동기의 "마법"

FastAPI는 개발자가 동기 코드와 비동기 코드를 자연스럽게 함께 사용할 수 있도록 놀라운 기능을 제공합니다.

• async def로 정의된 엔드포인트: FastAPI는 이를 메인 이벤트 루프에서 직접 실행합니다. I/O-bound 작업에 가장 효율적인 방식입니다.
• def로 정의된 엔드포인트: FastAPI는 이 일반적인 동기 함수를 별도의 스레드 풀(thread pool)에서 실행합니다. 이 덕분에 동기 함수가 오래 걸리더라도 메인 이벤트 루프를 막지(blocking) 않아 다른 요청 처리에 영향을 주지 않습니다.

하지만 가장 중요한 규칙이 있습니다.

🚨 async def 함수 안에서는 절대 이벤트 루프를 블로킹(blocking)하면 안 됩니다.

async def로 함수를 정의했다면, 그 안에서는 반드시 await를 사용하여 I/O 작업을 처리해야 합니다. 만약 time.sleep()과 같은 동기 블로킹 함수를 사용하면, 이벤트 루프 전체가 멈춰버려 비동기의 모든 장점이 사라지고 오히려 성능이 저하되는 재앙이 발생합니다.

# GOOD: 올바른 비동기 엔드포인트
@app.get("/async-non-blocking-sleep")
async def async_non_blocking_sleep():
    await asyncio.sleep(5) # 이벤트 루프 양보
    return {"message": "진정한 비동기"}

# BAD: 비동기 함수 내 블로킹 코드 (절대 금지!)
@app.get("/async-blocking-sleep")
async def async_blocking_sleep():
    time.sleep(5) # 이벤트 루프 블로킹! 서버 전체가 멈춤
    return {"message": "비동기 속의 재앙"}

---

3. SQLAlchemy를 비동기로 전환하기

이제 이론을 실전에 적용해 봅시다. 기존에 만들었던 동기 방식의 SQLAlchemy CRUD API를 완전한 비동기 방식으로 전환해 보겠습니다. 데이터베이스 연동은 대표적인 I/O-bound 작업이므로, 비동기 전환의 효과를 가장 크게 볼 수 있는 부분입니다.

Step 1: 비동기 DB 드라이버 설치 및 설정

먼저, SQLAlchemy가 비동기로 동작할 수 있도록 도와주는 라이브러리를 설치해야 합니다. SQLite를 사용하고 있으므로 aiosqlite를 설치합니다.

pip install aiosqlite greenlet

다음으로, 비동기 전용 데이터베이스 연결 설정을 만듭니다.

# connection_async.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# URL에 비동기 드라이버 명시: "sqlite+aiosqlite"
SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite:///./test.db"

# create_engine 대신 create_async_engine 사용
async_engine = create_async_engine(SQLALCHEMY_DATABASE_URL, echo=True)

# class_에 AsyncSession을 지정하여 비동기 세션 팩토리 생성
AsyncSessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=async_engine,
    class_=AsyncSession,
)

Step 2: CRUD API 비동기 전환

이제 각 API 엔드포인트를 async def로 변경하고, 데이터베이스 I/O가 발생하는 모든 지점에 await를 붙여줍니다.

Create (생성)

# crud.py
@router.post("/items/")
async def create_item(item: ItemCreate):
    async with AsyncSessionLocal() as session: # async with 사용
        db_item = Item(name=item.name, price=item.price)
        session.add(db_item) # 메모리 작업 (await 불필요)
        await session.commit() # DB IO 작업 (await 필수)
        await session.refresh(db_item) # DB IO 작업 (await 필수)
        return db_item

Read (조회)

비동기 조회 시에는 session.execute()로 DB에 쿼리를 실행하는 부분과, result.scalars().all()로 결과를 파이썬 객체로 변환하는 부분이 분리됩니다. I/O는 execute에서만 발생합니다.

# crud.py
@router.get("/items/")
async def get_items():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(Item)) # DB IO 작업 (await 필수)
        items = result.scalars().all() # 메모리 작업 (await 불필요)
        return items

Delete (삭제)

삭제 시에는 조회, 삭제, 커밋 세 단계 모두 DB와 통신할 수 있으므로 await를 붙여줍니다.

# crud.py
@router.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(Item).where(Item.id == item_id))
        db_item = result.scalars().first()
        if not db_item:
            raise HTTPException(status_code=404, detail="Item not found")

        await session.delete(db_item) # DB IO 작업 (await 필수)
        await session.commit() # DB IO 작업 (await 필수)
        return Response(status_code=204)

---

4. 코드를 아름답게: 의존성 주입 (Dependency Injection)

모든 API 함수마다 async with AsyncSessionLocal() as session: 코드를 반복해서 작성하는 것은 비효율적입니다. FastAPI의 의존성 주입(Dependency Injection, DI) 시스템을 사용하면 이 문제를 우아하게 해결할 수 있습니다.

DI는 함수가 필요로 하는 객체(의존성, 여기서는 DB 세션)를 외부에서 "주입"해주는 디자인 패턴입니다.

Step 1: 의존성 함수 생성

먼저, 세션을 생성하고 제공하는 역할을 하는 함수를 만듭니다. yield 키워드는 요청이 처리되는 동안 세션을 제공하고, 처리가 끝나면 세션을 자동으로 닫는 역할을 합니다.

# connection_async.py
from fastapi import Depends

async def get_async_session():
    async with AsyncSessionLocal() as session:
        yield session

Step 2: API에 의존성 주입

이제 API 함수의 매개변수로 세션을 선언하고, Depends를 사용하여 get_async_session 함수를 주입합니다.

# crud.py
from fastapi import Depends
from .connection_async import get_async_session

@router.post("/items/")
async def create_item(item: ItemCreate, session: AsyncSession = Depends(get_async_session)):
    # 이제 함수 내부에서 세션을 직접 생성할 필요가 없습니다!
    db_item = Item(name=item.name, price=item.price)
    session.add(db_item)
    await session.commit()
    await session.refresh(db_item)
    return db_item

이제 모든 API 함수에서 async with ... 구문을 제거하고 Depends로 세션을 주입받도록 리팩토링하면, 코드가 훨씬 깔끔해지고 테스트와 유지보수가 용이해집니다.

# crud.py
from connection_async import AsyncSessionFactory, get_async_session
from fastapi import FastAPI, HTTPException, status, Depends
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from connection import SessionFactory
from models import Item


app = FastAPI()


class ItemResponse(BaseModel):
    id: int
    name: str
    price: int


# C: 상품 등록 API
class ItemCreateRequest(BaseModel):
    name: str
    price: int


@app.post("/items", status_code=201)
async def create_item_api(
    body: ItemCreateRequest,
    session: AsyncSession = Depends(get_async_session),
) -> ItemResponse:
    new_item = Item(name=body.name, price=body.price)
    session.add(new_item)  # DB에 저장할 객체를 선별
    await session.commit()  # DB에 반영
    await session.refresh(new_item)
    return new_item


# R: 전체 상품 조회 API
@app.get("/items", status_code=200)
def get_items_api() -> list[ItemResponse]:
    with SessionFactory() as session:
        stmt = select(Item)  # statement = SQL 구문
        items = session.scalars(stmt).all()
        return items


# R: 단일 상품 조회 API
@app.get("/items/{item_id}", status_code=200)
async def get_item_api(item_id: int) -> ItemResponse:
    async with AsyncSessionFactory() as session:
        stmt = select(Item).where(Item.id == item_id)
        result = await session.execute(stmt)
        item: Item | None = await session.scalar(stmt)

        if item is None:
            raise HTTPException(
                status_code=404, detail=f"Item Not Found(id: {item_id})",
            )
        return item


# U: 상품 수정 API
class ItemUpdateRequest(BaseModel):
    name: str | None = None
    price: int | None = None


@app.patch("/items/{item_id}", status_code=200)
def update_item_api(item_id: int, body: ItemUpdateRequest) -> ItemResponse:
    with SessionFactory() as session:
        stmt = select(Item).where(Item.id == item_id)
        item: Item | None = session.scalar(stmt)

        if item is None:
            raise HTTPException(
                status_code=404, detail=f"Item Not Found(id: {item_id})",
            )

        # 객체의 값을 변경하고 commit()하면, 그대로 데이터가 DB에 반영됨
        if body.name:
            item.name = body.name
        if body.price:
            item.price = body.price

        session.commit()  # session에 등록된 데이터를 DB로 저장
        return item


# U: 상품 수정 API
class ItemReplaceUpdate(BaseModel):
    name: str
    price: int


@app.put("/items/{item_id}", status_code=200)
def replace_item_api(item_id: int, body: ItemReplaceUpdate) -> ItemResponse:
    with SessionFactory() as session:
        stmt = select(Item).where(Item.id == item_id)
        item: Item | None = session.scalar(stmt)

        if item is None:
            raise HTTPException(
                status_code=404, detail=f"Item Not Found(id: {item_id})",
            )

        item.name = body.name
        item.price = body.price

        session.commit()
        return item


# U: 상품 수정 API 비동기 버전
# @app.put("/items/{item_id}", status_code=200)
# async def replace_item_api_async(item_id: int, body: ItemReplaceUpdate) -> ItemResponse:
#     async with AsyncSessionFactory() as session:
#         stmt = select(Item).where(Item.id == item_id)
#         result = await session.execute(stmt)
#         item: Item | None = result.scalar()
#
#         if item is None:
#             raise HTTPException(
#                 status_code=404, detail=f"Item Not Found(id: {item_id})",
#             )
#
#         item.name = body.name
#         item.price = body.price
#
#         await session.commit()
#         return item


# D: 상품 삭제 API
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item_api(item_id: int) -> None:
    with SessionFactory() as session:
        stmt = select(Item).where(Item.id == item_id)
        item: Item | None = session.scalar(stmt)

        if item is None:
            raise HTTPException(
                status_code=404, detail=f"Item Not Found(id: {item_id})",
            )

        session.delete(item)
        session.commit()

# # D: 상품 삭제 API (비동기 버전 - 주석처리)
# @app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
# async def delete_item_api_async(item_id: int) -> None:
#     async with AsyncSessionFactory() as session:
#         stmt = select(Item).where(Item.id == item_id)
#         result = await session.execute(stmt)
#         item: Item | None = result.scalar()
#
#         if item is None:
#             raise HTTPException(
#                 status_code=404, detail=f"Item Not Found(id: {item_id})",
#             )
#
#         await session.delete(item)
#         await session.commit()

---

마무리하며

오늘은 FastAPI의 강력한 비동기 기능과 SQLAlchemy를 연동하는 방법을 깊이 있게 다뤘습니다.

• 비동기는 I/O 대기 시간을 효율적으로 사용하여 웹 애플리케이션의 성능을 극대화합니다.
• async def 안에서는 블로킹을 피하는 것이 무엇보다 중요합니다.
• SQLAlchemy의 비동기 ORM을 사용하면 데이터베이스 연동 코드까지 완전한 비동기로 만들 수 있습니다.
• 의존성 주입은 코드의 중복을 제거하고 유연성과 테스트 용이성을 높이는 필수 디자인 패턴입니다.

이 개념들을 잘 활용한다면, 어떤 복잡한 요구사항이라도 빠르고 견고하게 처리할 수 있는 고성능 FastAPI 애플리케이션을 구축할 수 있을 것입니다. 다음 시간에는 오늘 배운 내용을 바탕으로 실제 AI 모델을 서빙하는 실습을 진행해 보겠습니다. 기대해 주세요!