1. 프로젝트에 설치

앞서 배운 내용을 바탕으로 FastAPI 프로젝트에 SQLAlchemy를 적용해보겠습니다. 터미널에서 다음 명령을 실행하세요.

mkdir 04_2_db
cd 04_2_db
python -m venv venv
.\venv\Scripts\activate
pip install fastapi uvicorn sqlalchemy

mkdir 04_2_db
cd 04_2_db
python -m venv venv
.\venv\Scripts\activate
pip install fastapi uvicorn sqlalchemy

SQLite는 Python에 기본으로 포함되어 있으므로 별도로 설치할 필요가 없습니다.

2. 데이터베이스 설정

2.1 데이터베이스 URL 설정

DB를 연결할 수 있는 코드를 작성합니다. 보통은 database.py 파일을 생성하고 내용을 작성하지만 우리는 main.py 파일에 작성하겠습니다. 전체 코드는 마지막에 제공하니 일일이 붙여넣기를 하지 않아도 됩니다. 또한 모든 코드의 동작 원리를 이해하기 보다는 전체 코드의 맥락을 이해하고, 동작을 확인한 다음, 필요한 부분을 수정하거나 추가하는 것을 목표로 하세요.

# Database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

# Database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

여기서 SQLALCHEMY_DATABASE_URL은 SQLite 데이터베이스 파일의 경로를 지정합니다. sqlite:///./sql_app.db는 현재 디렉토리에 sql_app.db 파일을 생성합니다. 앞에 sqlite:///까지가 프리픽스 입니다. sqlite:///./sql_app.db와 sqlite:///sql_app.db는 같습니다. ./는 현재 폴더라는 의미입니다. 따라서 현재 디렉토리에 sql_app.db 파일이 생성됩니다.

engine은 연결 설정이라고 생각하시면 됩니다. create_engine 함수를 사용하여 데이터베이스와 연결을 설정합니다. 이 함수는 멀티 쓰레드 사용 등 다양한 옵션을 지원합니다.

SessionLocal은 데이터베이스 세션을 생성하는데 사용됩니다. 세션은 데이터베이스 연결을 나타내며, 각 코드는 독립적인 세션을 사용할 수 있습니다.

Base는 모델을 정의하는데 사용됩니다. 모델은 데이터베이스 테이블을 정의하는데 사용됩니다.

2.2 데이터베이스 세션 관리

main.py 세션을 관리할 수 있는 코드 입니다. 여기서 get_db는 FastAPI의 함수가 호출할 수 있도록(의존성 주입) 코드를 작성할 것입니다. 이렇게 호출된 함수는 데이터베이스 세션을 생성하고, 세션을 닫아주는 역할을 합니다. 세션을 닫아주지 않으면 데이터베이스 연결이 계속 유지되어 메모리 누수가 발생할 수 있습니다.

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

여기서 yield가 어떻게 사용되는지 예시를 통해 알아보겠습니다.

from fastapi import Depends
 
@app.get("/users/")
async def read_users(db: Session = Depends(get_db)):
    # 1. get_db()가 호출되어 새 세션 생성
    # 이제 db로 이 세션에 접근할 수 있음
    # 2. yield를 통해 이 세션이 여기로 전달됨
    users = db.query(User).all()
    return users
    # 3. 함수 실행이 끝나면 finally 블록이 실행되어 세션이 종료됨

from fastapi import Depends
 
@app.get("/users/")
async def read_users(db: Session = Depends(get_db)):
    # 1. get_db()가 호출되어 새 세션 생성
    # 이제 db로 이 세션에 접근할 수 있음
    # 2. yield를 통해 이 세션이 여기로 전달됨
    users = db.query(User).all()
    return users
    # 3. 함수 실행이 끝나면 finally 블록이 실행되어 세션이 종료됨

조금 더 어려운 말로 @app.get("/users/")를 라우트 핸들러라고 하는데 이 라우트 핸들러의 작업이 완료되면 finally 블록이 실행되어 세션을 종료하는 것입니다.

3. 모델 정의

모델은 데이터베이스 테이블을 정의하는데 사용됩니다. 보통은 models.py 파일을 생성하고 정의합니다. 우리는 main.py 파일에 작성하겠습니다.

# Model
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)
    price = Column(Float)

# Model
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)
    price = Column(Float)

여기서 Item은 데이터베이스 테이블을 정의하는데 사용됩니다. __tablename__은 테이블 이름을 지정합니다. id, name, description, price는 각각 컬럼을 정의합니다. Column은 컬럼을 정의하는데 사용되며, Integer, String, Float은 데이터 타입을 지정합니다. primary_key=True는 자동으로 증가되는 숫자값인 기본키를 지정합니다. index=True는 인덱스를 지정합니다.

# Create tables
Base.metadata.create_all(bind=engine)

# Create tables
Base.metadata.create_all(bind=engine)

모델을 정의했다면 데이터베이스를 생성해야 합니다. metadata.create_all를 하게 되면 실제 데이터베이스에 테이블이 생성됩니다.

4. Pydantic 스키마 정의

스키마라는 단어는 데이터베이스에서 사용되는 용어입니다. 데이터베이스의 테이블 구조를 정의하는데 사용됩니다. Pydantic 스키마는 데이터를 검증하고 파싱하는데 사용됩니다. 우리는 앞서 3-1 챕터에서 BaseModel을 사용하여 Pydantic으로된 스키마를 어떻게 사용하는지 배웠습니다. 보통은 schemas.py 파일을 생성하고 정의하지만 main.py 파일에 작성하겠습니다.

주의하셔야 할 것은 Pydantic으로 정의된 스키마는 ORM 모델과는 다릅니다. ORM 모델은 데이터베이스 테이블을 정의하는데 사용되고, Pydantic 스키마는 데이터를 검증하고 파싱하는데 사용됩니다. ORM 모델과 Pydantic 스키마는 서로 다른 역할을 가지고 있습니다. 따라서 이 두개의 모델은 변환을 해주어야 합니다.

# Pydantic schema
class ItemCreate(BaseModel):
    name: str
    description: str | None = None
    price: float
 
 
class ItemResponse(ItemCreate):
    id: int

# Pydantic schema
class ItemCreate(BaseModel):
    name: str
    description: str | None = None
    price: float
 
 
class ItemResponse(ItemCreate):
    id: int

5. CRUD 작업 구현

실제 CRUD 작업을 구현합니다.

# CRUD operations
@app.post("/items/", response_model=ItemResponse)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item
 
 
@app.get("/items/", response_model=list[ItemResponse])
def read_items(db: Session = Depends(get_db)):
    return db.query(Item).all()
 
 
@app.get("/items/{item_id}", response_model=ItemResponse)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

# CRUD operations
@app.post("/items/", response_model=ItemResponse)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item
 
 
@app.get("/items/", response_model=list[ItemResponse])
def read_items(db: Session = Depends(get_db)):
    return db.query(Item).all()
 
 
@app.get("/items/{item_id}", response_model=ItemResponse)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

여기서 처음 보는 코드가 나옵니다. db: Session = Depends(get_db) 코드입니다. 이 부분을 FastAPI의 의존성 주입이라고 합니다.

db: Session = Depends(get_db)는 두 부분으로 나눌 수 있습니다.

db: Session: 데이터베이스 세션 타입 명시
Depends(get_db): 의존성 주입 선언

여기서 Depends는 FastAPI에게 "이 함수가 실행되기 전에 먼저 get_db() 함수를 실행해서 그 결과를 db 매개변수로 주입해줘"라고 지시합니다. get_db() 함수를 다시 한 번 살펴보도록 하겠습니다. 여기서 SessionLocal()로 데이터베이스 연결을 하고 이 연결을 yield로 반환합니다. 이렇게 하면 FastAPI가 get_db() 함수를 실행하고 반환된 데이터베이스 세션을 db 매개변수로 주입해줍니다.

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

그리고 이 create_item()함수가 종료될 때 위 함수의 finally 블록이 실행되어 데이터베이스 세션이 종료됩니다. 이렇게 하면 데이터베이스 연결이 계속 유지되어 메모리 누수가 발생하지 않습니다.

이렇게 사용하게 되면 여러 엔드포인트에서 동일한 의존성을 쉽게 재사용할 수 있습니다. 여기서 의존성이라는 단어는 다른 함수나 클래스에서 사용하는 함수나 클래스를 의미합니다. 이렇게 하면 코드를 재사용하고, 중복을 줄일 수 있습니다.

6. 전체 코드

main.py 파일에 붙여넣고 실행해보세요.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
 
# Database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()
 
 
# Model
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)
    price = Column(Float)
 
 
# Create tables
Base.metadata.create_all(bind=engine)
 
 
# Pydantic schema
class ItemCreate(BaseModel):
    name: str
    description: str | None = None
    price: float
 
 
class ItemResponse(ItemCreate):
    id: int
 
 
app = FastAPI()
 
 
# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
 
# CRUD operations
@app.post("/items/", response_model=ItemResponse)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict()) # item을 dict로 변환하여 Item 모델로 생성
    db.add(db_item) # 데이터베이스에 추가(저장X, 세션에만 추가)
    db.commit() # 데이터베이스에 변경사항 저장
    db.refresh(db_item) # 데이터베이스에서 최신 정보로 업데이트
    # refresh는 동기화, 주로 전체 데이터를 다시 불러올 때 사용
    return db_item
 
 
@app.get("/items/", response_model=list[ItemResponse])
def read_items(db: Session = Depends(get_db)):
    return db.query(Item).all()
 
 
@app.get("/items/{item_id}", response_model=ItemResponse)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
 
# Database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()
 
 
# Model
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)
    price = Column(Float)
 
 
# Create tables
Base.metadata.create_all(bind=engine)
 
 
# Pydantic schema
class ItemCreate(BaseModel):
    name: str
    description: str | None = None
    price: float
 
 
class ItemResponse(ItemCreate):
    id: int
 
 
app = FastAPI()
 
 
# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
 
# CRUD operations
@app.post("/items/", response_model=ItemResponse)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict()) # item을 dict로 변환하여 Item 모델로 생성
    db.add(db_item) # 데이터베이스에 추가(저장X, 세션에만 추가)
    db.commit() # 데이터베이스에 변경사항 저장
    db.refresh(db_item) # 데이터베이스에서 최신 정보로 업데이트
    # refresh는 동기화, 주로 전체 데이터를 다시 불러올 때 사용
    return db_item
 
 
@app.get("/items/", response_model=list[ItemResponse])
def read_items(db: Session = Depends(get_db)):
    return db.query(Item).all()
 
 
@app.get("/items/{item_id}", response_model=ItemResponse)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

7. 애플리케이션 실행

이제 애플리케이션을 실행할 수 있습니다.

uvicorn main:app --reload

uvicorn main:app --reload

이제 FastAPI 애플리케이션이 SQLite 데이터베이스와 연동되어 실행됩니다. 아래 URL로 접속하여 API를 테스트해보세요.

http://127.0.0.1:8000/docs

썬더 클라이언트를 사용한다면 아래 URL로 접속하여 API를 테스트해보세요.

메서드: POST
URL: 127.0.0.1:8000/items

Body:

{
  "name": "item1",
  "description": "item1 description",
  "price": 100.0
}

{
  "name": "item1",
  "description": "item1 description",
  "price": 100.0
}

데이터가 들어갔다면 아래 URL로 접속하여 데이터를 확인해보세요.

메서드: GET
URL: 127.0.0.1:8000/items

8. 데이터 확인

데이터를 확인하는 방법으로는 SQLite 데이터베이스 파일을 직접 열어서 확인하는 방법이 있습니다. 더블클릭을 하면 바로 열리는 것은 아니기 때문에 SQLite Viewer 익스텐션을 설치하거나 DB Browser for SQLite 등의 프로그램을 사용하면 쉽게 확인할 수 있습니다.

연습문제

위의 CRUD 작업에 "update_item" 및 "delete_item" 함수를 추가해보세요.
새로운 모델 (예: "User")을 추가하고, 이에 대한 CRUD 작업과 API 엔드포인트를 구현해보세요.
Item 모델에 "created_at" 필드를 추가하고, 아이템 생성 시 자동으로 현재 시간이 저장되도록 구현해보세요.
SQLite 데이터베이스 파일의 경로를 환경 변수에서 읽어오도록 코드를 수정해보세요. (힌트: python-dotenv 라이브러리를 사용할 수 있습니다)

FastAPI 베이스캠프

FastAPI에 DB적용하기