🔥FastAPI MySQL 연동 및 CRUD까지

🕒 약 3분 읽는 데 소요됩니다

이전 글 보기(FastAPI 보안 OAuth2, JWT)

웹 애플리케이션에서 사용자의 정보, 게시글, 댓글, 상품 등 모든 데이터를 다루기 위해서는 안정적이고 효율적인 DB 연동이 필수다. 이번 포스팅에서는 FastAPI에서 MySQL과 SQLAlchemy를 사용하여 데이터베이스를 연동하고, CRUD(생성, 조회, 수정, 삭제) 기능을 구현하는 방법을 차근차근 설명하겠다. 개발 환경은 Windows, 데이터베이스는 MySQL, ORM은 SQLAlchemy를 사용한다.

---

⚙️ 환경 설정하기

먼저 필요한 패키지를 설치해야 한다. 아래 명령어를 터미널(cmd 혹은 PowerShell)에서 실행하자.

pip install fastapi[all] sqlalchemy pymysql

• sqlalchemy: ORM을 위한 핵심 라이브러리
• pymysql: MySQL 드라이버 역할
• fastapi[all]: FastAPI와 함께 Uvicorn 등 필요한 종속성 포함

MySQL 서버는 로컬 혹은 원격에 설치되어 있어야 하며, user, password, database, host, port 정보를 알고 있어야 한다.

---

🧩 DB 연결 설정 – database.py

다음은 SQLAlchemy의 엔진과 세션을 설정하는 코드이다.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# MySQL 접속 문자열
DATABASE_URL = "mysql+pymysql://username:password@localhost:3306/testdb"

engine = create_engine(DATABASE_URL, echo=True)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

여기서 username, password, testdb는 각자 환경에 맞게 바꾸어야 한다. echo=True는 SQL 로그를 출력해줘서 디버깅에 유용하다.

---

📦 모델 정의 – models.py

데이터베이스 테이블과 매핑되는 클래스를 SQLAlchemy로 정의하자.

from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)

이제 User 모델은 users라는 테이블로 변환되어 MySQL에 생성될 수 있다.

---

🚀 초기화 및 DB 테이블 생성

main.py 또는 별도의 초기화 스크립트에서 아래 코드를 실행하여 테이블을 생성하자.

from database import engine
from models import Base

Base.metadata.create_all(bind=engine)

이 코드를 통해 MySQL의 testdb 데이터베이스에 users 테이블이 생성된다.

---

🔨 CRUD 구현 – crud.py

ORM을 활용하여 데이터 조작 함수들을 구현해보자.

from sqlalchemy.orm import Session
from models import User

def create_user(db: Session, name: str, email: str):
    db_user = User(name=name, email=email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

def get_users(db: Session, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()

def update_user(db: Session, user_id: int, name: str):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        user.name = name
        db.commit()
        db.refresh(user)
    return user

def delete_user(db: Session, user_id: int):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
    return user

이제 기본적인 데이터 조작 작업은 모두 준비되었다.

---

🌐 FastAPI와 연결 – main.py

API 라우팅과 요청 처리 코드를 추가해보자.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal
from models import User
import crud
from pydantic import BaseModel

app = FastAPI()

# DB 세션 주입
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

📬 유저 생성

class UserCreate(BaseModel):
    name: str
    email: str

@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    return crud.create_user(db, name=user.name, email=user.email)

📥 유저 목록 조회

@app.get("/users/")
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud.get_users(db, skip=skip, limit=limit)

📤 유저 단건 조회

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

✏️ 유저 수정

class UserUpdate(BaseModel):
    name: str

@app.put("/users/{user_id}")
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    return crud.update_user(db, user_id, user.name)

❌ 유저 삭제

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    return crud.delete_user(db, user_id)

---

⚡ 비동기 DB 연동에 대해 잠깐

SQLAlchemy는 기본적으로 동기 방식이다. FastAPI는 비동기 웹 프레임워크이지만, SQLAlchemy와 함께 사용할 때는 주로 동기적으로 동작한다. 만약 비동기 DB 처리가 필요하다면, SQLModel + async SQLAlchemy, 혹은 Tortoise ORM 같은 라이브러리를 사용하는 것이 좋다.

비동기 처리를 하려면 async def 라우트와 await 호출이 필요하고, DB 세션도 async_sessionmaker로 처리해야 한다. 설정이 복잡해지는 대신 높은 처리 성능을 기대할 수 있다.

---

🔚 정리하며

이번 포스트에서는 Windows 환경에서 FastAPI와 MySQL을 연동하는 방법, 그리고 SQLAlchemy ORM을 통해 테이블 생성부터 CRUD 구현까지 전 과정을 알아보았다. FastAPI의 장점은 그 유연성과 확장성에 있다. 데이터베이스 연동 또한 그 중 하나다. 특히 실제 운영 환경에서는 데이터 무결성, 트랜잭션 처리, ORM 활용 능력이 중요하므로, 기본기를 탄탄히 다지는 것이 중요하다.

이제 여러분은 사용자 데이터를 DB에 저장하고 불러오며, 웹 서비스의 핵심 기능을 구축할 수 있게 되었다. 다음 단계에서는 인증 처리, 페이징, 검색 기능까지 확장해보자.

FastAPI 공식 문서 : https://fastapi.tiangolo.com/ko/