지식이 늘었다/웹개발

FastAPI - DB, SQLAlchemy와 ORM

PurpleGuy101 2025. 1. 5. 13:39

(0).

잡설 및 주절주절

 

백엔드 서비스를 배우는 많은 사람들이 

처음 예제로 접하는 프로젝트로 Todo List를 따라 만들어보는 과정을 가진다. 

이후 게시판 기능을 구현하면서 기초적인 백엔드 개발 과정이 끝난다.

 

단기 속성 과정에 가까워질수록 특정 프레임워크나 패키지의 사용법을 익히는 수준에 그치게 되기에

제대로 배우고 싶다면, 툴 사용법을 넘어서서 디테일한 부분을 찾아서 볼 필요가 있어보인다.

다만 그 양이 상당히 방대하여 패키지 딸깍만 해도 시간이 충분하지 않다..

그러므로 양적인 게임은 여전히 중요하다..

백엔드 강의의 흐름

0. 개발환경 설정 : IDE 설정, 의존성 관리 도구(pip, poetry 등)와가상환경 설정(virtualenv, conda 등), Git으로 버전 관리하기

1. API 엔드포인트 설계 및 구현: RESTful 디자인 원칙에 따라 API 설계

2. HTTP 메서드 처리: GET, POST, PUT, DELETE 기능 구현. Request와 Response의 데이터 직렬화/역직렬화 처리

3. 데이터베이스 연결: SQL/NoSQL 데이터베이스 모델링 설계, 서버와 연결. ORM(SQLAlchemy, Django ORM 등)을 활용

4. 인증(Authentication)과 권한 관리(Authorization): JWT, OAuth 2.0 같은 인증/권한 부여 방식 학습

5. 세션 및 토큰 관리: 세션 관리와 토큰 기반 인증(JWT 등) 구현

6. 동적 HTML 생성: 템플릿 엔진(Jinja2, EJS 등)을 활용한 서버사이드 렌더링(SSR)

7. AJAX 및 실시간 통신: AJAX와 fetch API를 통한 비동기 통신, WebSocket을 통한 실시간 기능 구현

8. 서드파티 라이브러리 사용: 외부 API(SDK 포함)를 활용한 서비스 통합

9. 형상 관리 및 배포: Git으로 버전 관리하고, 클라우드 서비스(AWS, GCP 등)나 Docker를 활용해 배포

10/ 테스트: 유닛 테스트, 통합 테스트, 자동화된 API 테스트(Pytest, Jest 등) 작성

 

 

(1)

 

강의를 보고 DB기능을 구현하였다. 

MySQL은 앞선 NodeJS 강의에서 사용해본적이 있어서 SQL 구문은 따로 필기하지 말자.

 

FastAPI에서 잡다한 DB보다 중요한 것은

SQLAlchemy 패키지같다.

SQLAlchemy로 DB를 설정하고, 연결하고, 객체를 저장하기 알맞은 형태로 다듬어 준 뒤,

add, delete, filter로 CRUD기능을 잘 만들어주면 기초공사가 끝난다.

 

MongoDB든 MySQL이든 잡다한 RDBMS들도

SQLAlchemy에서 잘 접합만 시켜주면 기존 코드의 변경없이 시스템을 유지보수하기 편해진다.

 

(2)

 

SQLAlchemy는 파이썬에서 많이 사용되는

'데이터베이스 접근' 라이브러리이다. 현 시점에서는 표준 ORM이라고 한다.

 

SQLAlchemy를 사용하고, database.py로 모듈을 분리하여

HTTP요청처리, Auth , DB 기능 부분으로 나눈후,

database.py에서 SQLAlchemy의 ORM(object-relational mapping)를 적절히 활용하도록 하자.

 

이러한 모듈화된 구조를 통해서 기존에 MySQL을 사용하다가도

Alembic을 사용하여 유연하게 Oracle이나 MongoDB로 Migration할 수도 있고

혹은 SQLite + SQLAlchemy를 통해 별도의 DB설치 없이 데이터를 저장하고 관리할 수도 있다. 

 

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

SQLALCHEMY_DATABASE_URL = 'sqlite:///./todosapp.db'
# create location on db on our fastapi application

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args = {'check_same_thread':False})
# define connection with db
# sql allow only one thread to communicate with
# preventing any kind of accident with different request

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

Base = declarative_base()

 

(4)

 

ORM이란?

 

(뭔 쉽게 설명할 것을 GPT에 넣으면 더 복잡해지는거 같냐)

 

ORM(Object-Relational Mapping)은

객체 지향 프로그래밍에서 사용하는 객체와

관계형 데이터베이스의 테이블 간의 구조적 차이를 연결하고 매핑하는 기술이다.

사용자가 애플리케이션에서 데이터를 요청(Request)하거나 저장(Save)하려고 할 때

백엔드에서는 이를 객체의 형태로 처리한다. 

객체란 데이터(속성)와 이를 다루는 메서드가 포함된 클래스를 기반으로 생성된 인스턴스를 의미하며, 

ORM은 이러한 객체를 관계형 데이터베이스의 테이블 구조에 맞게 변환한다.

관계형 데이터베이스에서는 스키마에서 정의된 엔티티(Entity)를 기준으로, 

각 속성(Attribute) 값들이 저장된 행(Row, Tuple)의 형태로 데이터가 저장된다. 

 

ORM은 객체의 속성과 테이블의 컬럼을 매핑하고, 

객체 인스턴스를 데이터베이스의 행으로 변환하여 저장하거나, 

데이터베이스에서 행을 조회하여 객체로 변환한다.

ORM은 이러한 과정을 통해 데이터베이스와 객체 지향 프로그래밍 간의 상호작용을 단순화하며, 

필요한 데이터만 선택적으로 저장하고 조회할 수 있도록 설계되어 있다.

(5)

 

GPT로 다듬으면 다듬을수록 글이 괜시리 길어지는 기분이다

 

한줄요약하자면

대충 SQLAlchemy로 DB를 FastAPI서버와 연결시켜준다는 점과

SQLAlchemy가 DB에 넣기 좋은 형태로 객체를 다듬어준다는게 핵심이다.

 

아래의 model.py를 통해

대충 SQLAlchemy에서 모델설정하고  DB연결하고

add, filter, delete로 CRUD해준다는 것만 알고 넘어가자.

 

#database.py

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

SQLALCHEMY_DATABASE_URL = 'sqlite:///./todosapp.db'
# create location on db on our fastapi application

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args = {'check_same_thread':False})
# define connection with db
# sql allow only one thread to communicate with
# preventing any kind of accident with different request

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

Base = declarative_base()

 

#model.py

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

class Todos(Base):
    __tablename__ = 'todos'

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    description = Column(String)
    priority = Column(Integer)
    complete = Column(Boolean, default=False)
#main.py 일부

@app.post("/todo", status_code = status.HTTP_201_CREATED)
async def create_todo(db: db_dependency, todo_request: TodoRequest):
    todo_model = Todos(**todo_request.dict())

    db.add(todo_model)
    db.commit()

 

 

 

# main.py 코드전문

from typing import Annotated

from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from fastapi import FastAPI, Depends, HTTPException, status, Path
import models
from models import Todos
from database import engine, SessionLocal

from routers import auth
# routers

app = FastAPI()

models.Base.metadata.create_all(bind = engine)
# 이 문구는, todos.db가 없을때만 실행될 것이다.
# 모델로 돌아가서 todos테이블을 향상시키면,
# 이 문구는 todos 테이블에 영향을 주지 않을 것이다. 

app.include_router(auth.router)

# DB dependency를 관리하자.
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

db_dependency = Annotated[Session, Depends(get_db)]

class TodoRequest(BaseModel):
    title: str = Field(min_length=3)
    description : str = Field(min_length=3, max_length=100)
    priority: int = Field(gt=0, lt=6)
    complete: bool
    

########################
# GET, POST, PUT, DELETE

@app.get("/", status_code=status.HTTP_200_OK)
async def read_all(db:Annotated[Session, Depends(get_db)]):
    return db.query(Todos).all()

@app.get("/todo/{todo_id}", status_code=status.HTTP_200_OK)
async def read_todo(db: db_dependency, todo_id:int = Path(gt=0)):
    todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
    # 일치하는 것을 찾는 순간 바로 반환
    if todo_model is not None:
        return todo_model
    raise HTTPException(status_code = 404, detail='Todo not found.')

@app.post("/todo", status_code = status.HTTP_201_CREATED)
async def create_todo(db: db_dependency, todo_request: TodoRequest):
    todo_model = Todos(**todo_request.dict())

    db.add(todo_model)
    db.commit()


@app.put("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(db: db_dependency, 
                    todo_id: int,
                    todo_request: TodoRequest):
    todo_model = db.qeury(Todos).filter(Todos.id == todo_id).first()
    if todo_model is None:
        raise HTTPException(status_code=404, detail='Todo not found.')
    
    todo_model.title = todo_request.title
    todo_model.description = todo_request.description
    todo_model.prioirity = todo_request.priority
    todo_model.complete = todo_request.complete
    
    db.add(todo_model)
    db.commit()

@app.delete("/todo/{todo_id}", status_code = status.HTTP_204_NO_CONTENT)
async def delete_todo(db: db_dependency, todo_id: int=Path(gt=0)):
    todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
    if todo_model is None:
        raise HTTPException(status_code=404, detail='Todo not found')
    db.query(Todos).filter(Todos.id == todo_id).delete()

    db.commit()

 

 

(기타)

 

Foreign Key

- 외래키는 관계형 데이터베이스의 테이블들을 연결하는 역할을 한다.

 

각각의 API요청에서, 유저는 고유한 ID를 배정attach받을 건데,

이 ID를 통해서 여러 테이블에서 이 유저와 관련된 정보를 찾아볼 수 있다.

유저에 관한 정보를 담은 여러개의 테이블들이

FK로 user ID를 가지고 있을 것이기 때문이다.

 

SELECT * FROM todos WHERE id=2

와 같이 DB에 쿼리를 날려서 레코드를 가져오는 과정을 수행해보자.