TIL/IoT Backend

FastAPI + SQLite 연동(센서 데이터 DB 저장)

masacorgi 2025. 6. 8. 23:26

이번 단계에선 FastAPI를 통해 받은 센서 데이터를 텍스트 로그가 아닌 실제 데이터베이스에 저장하는 구조로 확장하는 방법을 소개한다.
우리는 가볍고 설정이 간편한 SQLite를 사용하고, 추후 PostgreSQL로도 적용해보도록 하자.

1. 사용할 패키지 설치

pip install sqlalchemy fastapi uvicorn

sqlalchemy: Python ORM(Object Relational Mapper)
fastapi: 웹 프레임워크
uvicorn: 실행 서버

2. 전체 구조 개요

FastAPI ← POST /sensor
         ↓
    SQLAlchemy
         ↓
     SQLite DB

센서 데이터가 POST로 들어오면 SQLAlchemy가 자동으로 SQLite에 삽입해주는 구조이다.

3. DB 모델 정의 (models.py)

from sqlalchemy import Column, Integer, Float, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class SensorData(Base):
    __tablename__ = "sensor_data"

    id = Column(Integer, primary_key=True, index=True)
    device_id = Column(String)
    temperature = Column(Float)
    humidity = Column(Float)

4. DB 연결 및 초기화 (database.py)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./iot.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

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

5. FastAPI 연동 (main.py)

from fastapi import FastAPI, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session

from models import Base, SensorData
from database import engine, SessionLocal

# DB 테이블 생성
Base.metadata.create_all(bind=engine)

app = FastAPI()

# Pydantic 모델
class SensorSchema(BaseModel):
    temperature: float
    humidity: float
    device_id: str

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

@app.post("/sensor")
def create_sensor_data(data: SensorSchema, db: Session = Depends(get_db)):
    sensor = SensorData(
        device_id=data.device_id,
        temperature=data.temperature,
        humidity=data.humidity
    )
    db.add(sensor)
    db.commit()
    db.refresh(sensor)
    return {"status": "saved", "id": sensor.id}

6. 실행 방법

uvicorn main:app --reload

→ POST 요청으로 센서 데이터를 전송하면 자동으로 SQLite 파일(iot.db)에 저장돼.

7. DB 조회 방법 (추가 기능)

@app.get("/sensor")
def read_all_sensor_data(db: Session = Depends(get_db)):
    return db.query(SensorData).all()

→ 전체 센서 데이터를 JSON으로 확인할 수 있어.

8. 이번 후기

지금 우리는 단순한 데이터 수신 단계를 넘어서,
DB에 저장하여 분석·시각화·모니터링의 기반을 만드는 백엔드 구조를 만든 거야.
이제 데이터를 '한 번 받고 끝'이 아니라, 축적 가능한 자산으로 바꾼 셈이야.

SQLite는 작지만 강력하고,
PostgreSQL, MySQL로도 구조 그대로 옮기기 쉬운 설계라서 확장성도 좋아.