FastAPI-DB연결

FastAPI + SQLModel 기반 SQLite 데이터베이스 설정

데이터베이스 설정

FastAPI에서 SQLite를 기반으로 SQLModel을 사용한 데이터베이스 설정 방법을 정리한다. planner.db 파일이 프로젝트 루트에 생성되며, 앱 실행 시 자동으로 테이블이 생성되고 DB 세션이 관리된다.

# database/connection.py

from sqlmodel import SQLModel, create_engine, Session

# SQLite 데이터베이스 파일 경로 지정
database_file = "planner.db"
database_connection_string = f"sqlite:///{database_file}"
connect_args = {"check_same_thread": False}

# SQLite 데이터베이스 엔진 생성
engine_url = create_engine(
    database_connection_string,
    connect_args=connect_args,
    echo=True  # SQL 쿼리를 로그로 출력
)

# ✅ SQLModel 기반 테이블 자동 생성 함수
def conn():
    SQLModel.metadata.create_all(engine_url)

# ✅ FastAPI 의존성 주입용 DB 세션 생성기
def get_session():
    with Session(engine_url) as session:
        yield session

• planner.db: 프로젝트 실행 시 생성되는 SQLite DB 파일
• SQLModel.metadata.create_all(): 정의된 모델 기반으로 테이블 자동 생성
• get_session(): FastAPI 라우터에서 DB 세션을 주입할 때 사용되는 제너레이터 함수

모델 정의 요약

SQLModel을 상속받아 테이블 구조를 정의하며, tags 필드는 JSON 형태로 저장 가능하다.

# models/events.py

from typing import List, Optional
from sqlmodel import Field, SQLModel, Column, JSON

class Event(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    title: str
    image: str
    description: str
    tags: List[str] = Field(sa_column=Column(JSON))  # JSON 배열로 저장
    location: str

class EventUpdate(SQLModel):
    title: Optional[str] = None
    image: Optional[str] = None
    description: Optional[str] = None
    tags: Optional[List[str]] = None
    location: Optional[str] = None

애플리케이션 시작 시 DB 연결 및 테이블 생성

FastAPI에서는 앱의 생명주기 동안 실행할 작업을 lifespan 함수로 관리할 수 있다. 이 구조를 활용하면 서버 시작과 동시에 테이블이 생성되며, 종료 시 후처리도 가능하다.

# main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from routes.users import user_router
from routes.events import event_router
from database.connection import conn

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("애플리케이션 시작")
    conn()  # ✅ DB 연결 및 테이블 생성
    yield
    print("애플리케이션 종료")

app = FastAPI(lifespan=lifespan)

app.include_router(user_router, prefix="/users")
app.include_router(event_router, prefix="/events")

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

@asynccontextmanager

• Python의 비동기 컨텍스트 매니저를 정의하기 위한 데코레이터
• FastAPI(lifespan=...) 구조와 함께 사용하여 앱 실행 전후 로직을 지정
• yield 이전은 시작 시 실행, 이후는 종료 시 실행
• 이전 방식인 @app.on_event("startup")보다 최신 방식

FastAPI + SQLModel: DB 연동 CRUD 코드 및 주요 함수 정리

---

1. 이벤트 등록 - POST /events

@event_router.post("/", status_code=status.HTTP_201_CREATED)
async def create_event(data: Event, session: Session = Depends(get_session)) -> dict:
    session.add(data)
    session.commit()
    session.refresh(data)
    return {"message": "이벤트 등록이 완료되었습니다."}

• session: Session = Depends(get_session) → DB 연결 세션을 FastAPI 의존성 주입으로 받음
• session.add(data) → 전달받은 데이터를 DB에 INSERT 예정으로 추가
• session.commit() → 실제로 DB에 반영
• session.refresh(data) → DB 반영 이후 ID 등 자동 생성 필드를 다시 읽어옴

---

2. 이벤트 전체 조회 - GET /events

@event_router.get("/", response_model=List[Event])
async def retrieve_all_events(session: Session = Depends(get_session)) -> List[Event]:
    statement = select(Event)
    results = session.exec(statement)
    return results.all()

• select(Event) → SQLModel을 통한 SELECT 쿼리 작성
• session.exec(...) → 쿼리 실행
• .all() → 결과 전체를 리스트로 반환

💡 .all()을 사용하지 않으면 Result 객체 자체가 반환되어 응답 직렬화에 실패할 수 있음

---

3. 이벤트 상세 조회 - GET /events/{event_id}

@event_router.get("/{event_id}", response_model=Event)
async def retrieve_event(event_id: int, session: Session = Depends(get_session)) -> Event:
    event = session.get(Event, event_id)
    if not event:
        raise HTTPException(status_code=404, detail="일치하는 이벤트를 찾을 수 없습니다.")
    return event

• session.get(Event, id) → 기본키 기준 단일 객체 조회에 최적화
• 결과가 없으면 404 예외 반환

---

4. 이벤트 수정 - PUT /events/{event_id}

@event_router.put("/{event_id}", response_model=Event)
async def update_event(
    data: EventUpdate,
    event_id: int = Path(...),
    session: Session = Depends(get_session)
) -> Event:
    event = session.get(Event, event_id)
    if not event:
        raise HTTPException(status_code=404, detail="일치하는 이벤트를 찾을 수 없습니다.")

    update_data = data.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(event, key, value)

    session.add(event)
    session.commit()
    session.refresh(event)

    return event

• exclude_unset=True → 요청에서 실제 전달된 필드만 딕셔너리로 변환
• update_data.items() → 수정할 필드명과 값을 반복
• setattr(event, key, value) → 해당 필드만 동적으로 업데이트

💡 일부 필드만 전달된 경우에도 안전하게 처리 가능 (예: 제목만 수정)

---

5. 이벤트 하나 삭제 - DELETE /events/{event_id}

@event_router.delete("/{event_id}")
async def delete_event(event_id: int, session: Session = Depends(get_session)) -> dict:
    event = session.get(Event, event_id)
    if not event:
        raise HTTPException(status_code=404, detail="일치하는 이벤트를 찾을 수 없습니다.")
    session.delete(event)
    session.commit()
    return {"message": "이벤트 삭제가 완료되었습니다."}

• session.delete(event) → 해당 객체 삭제 예약
• session.commit() → 삭제 쿼리 DB 반영

---

6. 이벤트 전체 삭제 - DELETE /events

@event_router.delete("/")
async def delete_all_events(session: Session = Depends(get_session)) -> dict:
    statement = select(Event)
    results = session.exec(statement).all()
    for event in results:
        session.delete(event)
    session.commit()
    return {"message": "이벤트 전체 삭제가 완료되었습니다."}

• 전체 데이터를 먼저 조회하고 하나씩 삭제
• session.exec(text("DELETE FROM event"))으로 한 번에 삭제도 가능 (단, 비추천 시 주의)