[Python] FastAPI 비동기 처리(2) 비동기 데이터베이스 연결 구현하기

안녕하세요.

이전 글에서 FastAPI 비동기 처리를 위해 비동기 엔드포인트 만드는 방법을 알아보았습니다.

2025.05.15 - [Python] - [Python] FastAPI 비동기 처리(1) 비동기 엔드포인트 만들기

[Python] FastAPI 비동기 처리(1) 비동기 엔드포인트 만들기

 

FastAPI의 비동기 기능을 제대로 활용하려면 데이터베이스와의 연결도 비동기로 처리하는 것이 중요합니다.

 

이번 글에서는 ORM의 개념, 주요 비동기 ORM 라이브러리, 그리고 SQLAlchemy를 이용한 비동기 데이터베이스 연동 방법과 CRUD 예제를 다룹니다.

 

1. ORM(Object-Relational Mapping)이란?

FastAPI에서 데이터베이스를 효율적으로 다루기 위해서는 ORM이라는 개념을 이해해야 합니다.

 

ORM(Object-Relational Mapping, 객체-관계 매핑)은 프로그래밍 언어의 객체(클래스, 인스턴스)와 데이터베이스의 테이블, 행(row), 열(column)을 자동으로 연결해주는 도구입니다.

즉, SQL 쿼리를 직접 작성하지 않고, 파이썬 객체를 다루듯이 데이터베이스 작업을 할 수 있게 해줍니다.

 

예시

SQL 직접 사용

cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("홍길동", "hong@test.com"))

 

 

ORM 사용

user = User(name="홍길동", email="hong@test.com")
db.add(user)
db.commit()

ORM의 장점

• SQL을 몰라도 데이터베이스를 쉽게 다룰 수 있다.
• 코드가 더 깔끔하고 유지보수도 쉽다.
• 보안에 유리하다(SQL Injection 방지 등)

ORM의 단점

• 아주 복잡한 쿼리는 직접 SQL을 쓰는 것보다 비효율적일 수 있다.
• 변환 과정에서 약간의 성능 저하가 있을 수 있다.

FastAPI에서는 대표적으로 SQLAlchemy라는 ORM을 많이 사용합니다. 

2. SQLAlchemy를 활용한 비동기 DB 연결 실습

2-1. 환경 준비

비동기 데이터베이스 연동을 위해 다음과 같은 패키지를 설치해야 합니다.

pip install fastapi[all] sqlalchemy aiosqlite

• fastapi[all]: FastAPI와 Uvicorn 등 필수 패키지 포함
• sqlalchemy: ORM 라이브러리
• aiosqlite: SQLite의 비동기 드라이버
  (PostgreSQL을 쓸 경우 asyncpg를, MySQL은 aiomysql을 사용)

2-2. 비동기 엔진 및 세션 만들기

SQLAlchemy의 비동기 엔진과 세션을 아래와 같이 설정합니다.

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(DATABASE_URL, echo=True)
SessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)
Base = declarative_base()

• create_async_engine: 비동기 데이터베이스 엔진 생성
• AsyncSession: 비동기 세션 클래스
• declarative_base: ORM 모델 생성을 위한 베이스 클래스

2-3. 모델 정의

ORM에서 사용할 테이블 구조를 파이썬 클래스로 정의합니다.

from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

 

2-4. 데이터베이스 테이블 생성

초기 테이블을 생성하려면 아래와 같이 실행합니다.

import asyncio

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# 최초 1회만 실행
# asyncio.run(init_db())

 

3. 비동기 세션 관리 방법

FastAPI에서는 의존성 주입(Dependency Injection) 기능을 활용해 DB 세션을 안전하게 관리할 수 있습니다.

from fastapi import Depends

async def get_session() -> AsyncSession
    async with SessionLocal() as session:
        yield session

• 이 함수는 요청이 들어올 때마다 새로운 세션을 생성하고,요청이 끝나면 자동으로 세션을 정리합니다.

4. CRUD 예제

이제 실제로 데이터를 추가(Create)하고, 조회(Read)하는 API 엔드포인트를 만들어보겠습니다.

from fastapi import FastAPI, HTTPException
from sqlalchemy.future import select

app = FastAPI()

# 사용자 생성
@app.post("/users/")
async def create_user(name: str, email: str, session: AsyncSession = Depends(get_session)):
    new_user = User(name=name, email=email)
    session.add(new_user)
    await session.commit()
    await session.refresh(new_user)
    return new_user

# 사용자 전체 조회
@app.get("/users/")
async def read_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    users = result.scalars().all()
    return users

• create_user: 새로운 사용자를 추가합니다.
• read_users: 등록된 모든 사용자를 조회합니다.