FastAPI 베이스캠프

SQLAlchemy와 SQLite 연동

1. SQLite 데이터베이스 연동 소개

실제 애플리케이션에서는 데이터를 영구적으로 저장하고 효율적으로 관리하기 위해 데이터베이스를 사용합니다. 이 챕터에서는 FastAPI 애플리케이션에 SQLite 데이터베이스를 연동하는 방법을 알아보겠습니다. SQLite는 파일 기반의 경량 데이터베이스로, 별도의 설치 없이 바로 사용할 수 있기에 채택하였습니다. 만약 실무에서 사용할 경우 PostgreSQL, MySQL 등의 데이터베이스를 사용하는 것이 일반적입니다.

2. 필요한 라이브러리 설치

먼저, 필요한 라이브러리를 설치해야 합니다. 터미널에서 다음 명령을 실행하세요.

pip install sqlalchemy

pip install sqlalchemy

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

3. 데이터베이스 설정

3.1 데이터베이스 URL 설정

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

# Database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

여기서 SQLALCHEMY_DATABASE_URL은 SQLite 데이터베이스 파일의 경로를 지정합니다. sqlite:///./sql_app.db는 현재 디렉토리에 sql_app.db 파일을 생성합니다.

3.2 데이터베이스 세션 관리

main.py 파일에 데이터베이스 세션을 관리할 수 있는 코드를 작성합니다. 세션은 데이터베이스 연결을 나타내며, 데이터베이스 작업이 끝나면 세션을 닫아야 합니다. 만약 세션을 닫지 않으면 데이터베이스 연결이 계속 유지되어 메모리 누수가 발생할 수 있습니다.

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

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

4. 모델 정의

모델은 데이터베이스 테이블을 정의하는데 사용됩니다. 보통은 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)

모델을 정의했다면 데이터베이스를 생성해야 합니다. 이를 위해 다음 명령어가 뒤에 이어집니다.

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

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

이렇게 하면 모델에 따라 테이블이 생성됩니다.

5. Pydantic 스키마 정의

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

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

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

6. 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

7. 전체 코드

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, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, 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
 
    class Config:
        orm_mode = True
 
 
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, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, 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
 
    class Config:
        orm_mode = True
 
 
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

8. 애플리케이션 실행

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

uvicorn main:app --reload

uvicorn main:app --reload

위 명령어를 반복해서 실행해야 한다면 아래와 같이 수정할 수도 있습니다.

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

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

맨 마지막줄에 위 코드를 삽입하면 아래 명령어로도 실행할 수 있습니다.

python main.py

python main.py

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

http://127.0.0.1:8000/docs

9. 데이터 확인

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

연습문제

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