Fast API - SQL 데이터베이스

Programming Language/Python

Fast API - SQL 데이터베이스

hustle_D 2025. 9. 15. 15:29

Fast API에서 DB 를 사용하기 위해서 SQLModel을 사용해보도록 하자.

먼저 SQL Model의 경우는 SQL Alchemy와 Pydantic을 기반으로 구축되어 있는 모듈이다.

먼저 SQL Model을 설치해보자.

그러고 이제 모델을 생성하기 위해 필요한 모듈들을 추가해주고

Class로 모델을 생성해주자.

1. Model 생성

이때 SQLModel을 상속하면서 table 속성을 True로 설정해주자.

그리고 이안에 id, name, age, full_name을 멤버로 만들어주자.

이는 Pydantic 모델을 생성하는 방법과 매우 유사하게 사용된다.(내부적으로는 pydantic모델이긴하다고 함)

그 상세를 보면 몇가지 차이점이 있는데 먼저 prymary_key로 설정되어 있는 부분은 해당 멤버가 DB에서는 pk로써 사용될 것임을 SQL Model에게 알리는 것이다

또한 Annotated[str, None]이라면 해당 컬럼의 경우는 Nullable이라는 것을 알 수 있고 default가 None으로 잡히는 것을 보면 기본 값은 None으로 들어갈 것임을 알 수 있다.

또한 index=True로 설정된 것은 해당 컬럼에 대해서 SQL 인덱스를 생성하도록 지시하는 것이다.

2. Engine 생성

DB를 연결하기 위해서는 DB의 종류, DB에 맞는 Driver종류, username, password, host:port, database_name의 데이터를 가지고 있어야 한다.

그 설정은 각각

아래와 같이 문자열로 구성하면 된다.

DB_type+DB_driver://username:password@host:port/database_name

그리고 create_engine이란 함수에게 해당 문자열을 전달해주면 해당 DB를 통괄하는 엔진을 반환한다.

이제 해당 엔진을 사용해서 사용자의 접속에 대한 세션을 생성하거나 제거하는 역할을 한다.

3. 테이블의 생성

이제 함수를 하나 생성해서 아래와 같이 코드를 생성해주자.

각 코드의 설명은

SQLModel.metadata : 우리가 위에서 SQLModel을 상속해서 클래스(모델)을 만드는 경우 SQLModel.metadata라는 곳에 모이게 되고 그 모델에 대한 정보에 대한 처리를 위한 객체이다.
.create_all(engine) : 이 메더스가 실제 작업을 수행하는 부분으로 인자로 전달한 engine을 통해서 데이터베이스에 연결하고 SQLModel.metadata에 저장된 모든 테이블 설계 정보(우리가 생성한 SQLModel을 상속한 Model)를 확인하고 설계도에 따라서 CREATE TABLE SQL 명령을 실행해서 물리적인 테이블을 생성한다.

4. DB 세션 생성

DB는 접속하는 사용자마다 각각 세션을 생성하고 DB 작업을 진행한 후에 해당 세션을 종료시켜줘야 한다.

이 부분을 함수하나로 생성해서 호출하면 세션을 생성해주고 호출한 함수가 종료되면 세션을 제거하는 함수를 하나 생성해주자.

여기서 Session()은 SQLModel/SQLAlchemy의 동기 세션클래스로 인자로 전달한 engine에 해당하는 DB에 새로운 세션 인스턴스를 생성하고 열어준다.

그리고 with문으로 열었기 때문에 yield session이 끝나고 나면 혹은 exception으로 종료되더라도 session을 close해준다.

추후 해당 함수는 Depends의 인자로 전달되어 DB 세션을 엔드포인트에게 의존성 주입으로 전달하게 될것이고 해당 엔드포인트 메서드가 종료된다면 해당 세션의 yield가 끝나게 되면서 자동으로 sesison을 close할 게 될 것 이다.

또한 이렇게 생성된 세션을 사용해서 엔드포인트는 DB와의 상호작용을 하게될 것이다.

5. DB 세션

이제 메서드를 만들어서 테이블을 조작할 수 있도록 할것인데 기본적으로 각각의 메서드에서는 DB에 접근하기 위해서 엔진에 세션을 요청하고 이를 세션 객체로 전달해서 메서드가 DB작업을 진행한 다음에 완료되면 메서드가 종료되고 세션을 끊는 형태로 구성된다

여기서 세션을 생성하는 것은 외부로 뽑아내어 의존성 주입의 형태로 세션을 주입받게 되는데 그렇기에 함수의 형태는

와 같은 형태를 띄게 된다.

근데 결국 세션을 전달받는 방식은 동일한 함수에 동일한 세션의 형태로 주입을 받을 것이기에 이를 외부로 빼서 변수로 저장한 후에

이렇게 해서 각각의 메서드는 단순하게 개발 할 수 있도록 변수에 세션을 주입하는 코드를 담아 뒀다.

이제 예시를 보기 위해서 생성했던 get_user를 제거해주자

6. Table 생성

이제 위 코드를 사용해서 DB에 테이블을 생성해보도록 하자.

먼저 fastapi의 서버를 만들기 전에 이제 함수를 생성할 건데 먼저 데코레이터 하나를 보고 가자

@app.on_event("startup")

~~@app.on_event("startup")~~은 FastAPI 프레임워크에서 사용되는 이벤트 핸들러 데코레이터이다.

해당 데코레이터는 FastAPI 애플리케이션이 시작될 때 특정 코드를 실행할 수 있도록 함수를 등록하는 역할을 한다.

그래서 보통은 데이터베이스의 연결(서버가 시작될 때 데이터베이스에 연결하고 종료될때 연결을 끊는 작업을 수행할 수 있다)

리소스 로드(머신러닝 모델, 설정 파일, 캐시 데이터 등 애플리케이션이 실행되기 전에 미리 로드해야하는 리소스를 불러온다)

초기화 작업(서버가 요청을 받기 전에 필요한 초기화 작업을 실행한다(로깅 시스템 설정, 외부 API 연결등)) 등에 사용된다.

lifespan

@app.on_event("startup")이 deprecated되면서 대체 방식으로 lifespan이란것을 사용하는 것으로 보인다.

lifespan은 FastAPI 애플리케이션의 시작과 종료 시점에 특정 코드를 실행하도록 돕는 기능이다.

이는 데이터베이스의 연결, 외부 리소스 로드, 캐시 초기화등 애플리케이션의 생명 주기와 관련된 작업을 안전하게 처리하도록 하기 위해 도입되었다.

lifespan은 비동기 컨텍스트 관리자(asynchronous context manager)를 사용해서 구현된다.

파이썬의 with 구문과 비슷하게, 진입 시점과 종료 시점을 명확하게 구분해서 코드를 실행한다.

먼저 lifespan을 사용하려면 contextlib의 asynccontextmanager 데코레이터를 사용해서 함수를 정의하고 이를 FastAPI에게 전달해주면 된다.

이제 내부에 시작시 실행될 코드, 종료시 실행될 코드로 분리해서 작성하면된다.

그리고 그 기준은 yield으로 yield이전은 프로그램이 시작될때 yield이후는 프로그램이 종료될때 실행될 코드를 작성하면 된다.

이렇게 만든 lifespan을 이제 FastAPI의 생성자에게 전달해주면 서버가 시작, 종료될때 해당 함수를 보고 수행을 하게 된다.

7. 데이터 Insert

이제 생성한 테이블에 데이터를 insert해주는 메서드를 만들어보자.

해당 메서드는 엔드포인트로 작동할 것인데 해당 엔드포인트에 접근할때 가지고올 데이터를 갖고 데이터를 insert하도록 만들어 줄것이다.

먼저 함수를 하나 생성해주고

이제 여기서 파라미터로 ModelTest에 맞는 데이터를 받을 수 있도록 하고

세션을 주입받자

이제 이 세션을 통해서 작업을 할건데 이 때 사용할 session의 메서드를 각각 살펴보면

1) session.add()

add 함수는 객체를 세션에 추가하는데 이 함수를 호출해도 데이터베이스에 실제 데이터가 삽입되는 것은 아니다.

이는 물건을 장바구니에 담아두는것 처럼 새로운 데이터, 변화된 데이터에 대해서 세션이 추적하고 관리하도록 하게 해준다.

이 함수는 insert, update시에 사용된다

2) session.commit()

commit 함수는 세션에 추적되던 모든 변경 사항을 데이터베이스에 반영하고 트랜젝션을 종료시킨다.

이 시점에서 insert, update, delete등의 SQL명령이 실행되고 데이터베이스의 내용이 실제로 바뀐다.

3) session.refresh()

refresh 함수는 데이터베이스의 최신 상태로 객체를 갱신한다.

이 함수는 주로 다른 곳에서 데이터가 변경되었을때 사용된다.

세션에 있는 객체는 변경사항이 없더라도 데이터베이스의 최신 상태와 다를 수 있는데 refresh는 데이터베이스를 다시 조회하며 객체의 속성을 최신값으로 업데이트 한다.

그래서 사용할때 add -> commit -> refresh 를 세트로 사용하게 된다.

각각 순서대로 사용해주면

이렇게 작성할 수 있다.

8. 데이터 Select

이제 데이터를 가져오는 방법을 살펴볼 것이다.

사실 여기서는 크게 어려울건 없고 세션을 주입받은 다음에 세션을 통해서 데이터를 조회하면 된다.

단건의 경우 세가지 방식, 다건일 경우 두가지 방식이 사용 가능한데

먼저 단건에 대해서 확인해보도록하자

단건 Select

먼저 단건을 조회하는 함수를 하나 만들어주고

이제 데이터를 받아오는 방식을 여러개로 나눠서 확인해볼 텐데 가장 먼저

1) session.get()

session.get의 기본적인 사용방법은 아래와 같다.

return session.get(ModelTest, id)

위와 같은 방법으로 사용이 가능하다

이 경우는 select modeltest.* from modeltest where id = :id와 동일한 기능을 한다

이 때는 pk를 사용해서 조회될 때만 가능한 방식으로 가장 간단한 단건 조회 방식이다.

이 방식을 사용해서 메서드를 만든다면

이렇게 메서드를 만들 수 있다.

2) session.exec()

session.exec()의 기본적인 사용방법은 아래와 같다

stmt = (
    select(ModelTest)               # SELECT modeltest.* FROM modeltest
    .where(ModelTest.id == id)		# WHERE id = :id
)
    
return session.exec(stmt).first()

쿼리를 만드는 방식은 동일하나 값을 가져오는 방식은 first()를 사용해서 가져온다

이때는 2개 이상 나올 경우 하나만 반환 되며 에러를 내지는 않는다.

해당 방식을 사용한다면

이렇게 메서드를 만들 수 있다.

## 원래 session.query(), session.execute()를 사용하는 방식도 있었는데 deprecated 되었다는듯...

다건 select

다건의 경우는 pk를 사용해서 단건 조회 하던 방식을 빼고 exec를 사용하면 된다.

다만 이때 조건을 여러개 붙일 수 있는데

stmt = (
    select(ModelTest)               # SELECT modeltest.* FROM modeltest
    .where(ModelTest.id == id)		# WHERE id = :id
    .order_by(ModelTest.id.asc())	# ORDER BY id ASC
    .offset(skip)					# OFFSET :skip
    .limit(limit)					# LIMIT :limit
)
    
return session.exec(stmt).all()

이렇게 조회하면 list[ModelTest]타입으로 값을 반환한다.

다건조회 하는 경우에는 이런 저런 조건들을 붙일 수 있다.

보면 where 로 조회 조건을 걸고 order_by를 통해서 조회 결과의 정렬을 맞추고 offset으로 어디서 부터 데이터를 읽을지 시작점을 지정하고 limit를 통해서 몆건의 데이터만을 불러들일것인지 설정을 하는 등의 설정이 가능하다.

9. 데이터 Delete

데이터의 delete도 크게 다르지 않다

먼저 함수를 하나 생성해주고

먼저 삭제하고자 하는 데이터를 조회한다.

이때는 내가 원하는 방향이 하나만 제거하는 것이라면 2개 이상인 경우에는 에러를 벹어줘야 한다.

또한 값이 없는 경우도 DB를 commit()해서 작업을 진행할 필요가 없기 때문에 그전에 exception을 던져줘야 한다.

일단 그런 후에 session의 delete() 함수를 사용하면

삭제하고자 하는 raw를 전달하면 ORM엔티티에 삭제를 예약하는 것이다.

이러고 DB에 반영하기위해서 commit()을 해서 실제 데이터를 delete해주자.

refresh를 하지 않는 이유는 refresh는 DB에 아직 존재하는 행을 다시 읽어와 ORM 객체 필드에 최신화하는 동작인데 delete후 commit하면 그 행은 사라지기 때문에 새로고침할 대상이 없어 refresh하면 에러를 벹는다.

10. 서버 실행 및 DB 테스트

먼저 서버가 켜질 수 있도록 코드를 추가하고

서버를 실행해보자.

여기서 서버 리로드를 할 경우 테이블이 존재하는 경우는 테이블을 생성을 만들지 않도록 테이블 설정하는 부분에

이렇게 설정 추가해주고 시작해주자.

## 내부에 구체적으로 설명하지 못했는데 하다보니 조금 이런 저런 오류들이 있었다.

지금 저 프로젝트 자체는 로컬에 있는게 맞으나 저걸 띄우는 bash가 wsl에 존재하기 때문에 DB 서버에 접근하기 위해서는 외부 DB에 접근하듯이 사용해야 한다는 문제점으로 db주소를 원래는 localhost로 설정했으나 접근이 안돼서 그냥 wsl 내부에서 docker로 postgresql을 띄우고 로컬호스트로 접근하는 방법으로 변경했다(위 설명 부분에서는 따로 언급한 내용이 아니기에 크게 영향이 없을 수도 있음)

#WSL과 Windows의 관계#

Windows 에서 claude를 사용하면서 wsl의 bash를 사용할 수 밖에 없는 상황이라 bash를 사용하는데 애플리케이션 개발 중 DB 접근을 할때 Windows 의 local에 띄워둔 DB에 wsl로 실행시킨 애플리케이션이 접근을 하지 못하는 상황이 발생해서 이런 저런 방식을 사용하다가 WSL에서 Windows로 접근할때에는 host와 port를 찾아서 접근해야하는 상황인데 Windows에서 WSL로 접근할때에는 localhost로 그냥 접근이 가능한 경우를 보면서 조금 정리가 필요할듯 싶어 내용을 작성한다.

Windows → WSL

Windows에서 WSL로 localhost:port로 접근하는 경우는 문제가 없이 가능하다.

그 이유는 WSL2은 기본값으로 localhost에 대해서 forwarding을 켜둬서 Windows의 127.0.0.1:port로 접근한 트래픽을 WSL VM의 ip:port로 프록시 해준다.(WSL이 호스트에 포트 프록시를 자동 설치)

이 때 서비스는 0.0.0.0(또는 ::)로 리슨하고 있어야만 한다

(0.0.0.0으로 리슨 하고 있다는 의미는 IPv4에 대해서 외부 접근을 허용한다는 의미로 127.0.0.1로 리슨하고 있는 경우 루프백만 수신하는 것으로 실제 WSL의 내부에서 127.0.0.1로 리슨하면 Windows에서는 접근이 불가능하다.

그리고 ::으로 리슨하는것도 IPv6에 대해서 외부 접근을 허용한다는 의미로 IPv6의 경우는 ::1으로 루프백만 수신한다고 설정이 가능하고 따로 숫자없이 ::만 사용하면 외부 모든 IPv6에 대해서 접근을 허용한다는 의미가 된다)

WSL → Windows

WSL에서 Windows로 localhost:port로는 그냥 접근이 불가능하다

이는 WSL 안에 localhost는 WSL 자신을 뜻하는 것이고 WSL → Windows의 자동 프록시는 없기 때문에 Windows 호스트의 IP로 접근해야만 한다.

#__table_args__ = = {'extend_existing': True} 와 SQLModel.metadata.create_all(engine, checkfirst=True) #

작업을 진행하다가 테이블이 중간에 없어지는 문제가 발생했다.

이유는 정확하게 모르겠으나 그 상태로 키는건 문제가 없는데 index가 이미 존재한다고 계속 나오는 상황에서 에러가 잡히질 않았다.

그래서 새로 추가된 부분이 checkfirst=True 부분인데 이걸 설정했더니 바로 해결 됐다.

나는 기존에 extend_existing=True로 설정하면 기존에 이미 테이블이 존재할때 에러를 방지하는 것으로 생각했었는데 그게 또 아니였던 것 같아서 추가로 두 부분에 대해서 이해하고 넘어가야할듯 싶다.

__table_args__ = {'extend_existing': True}

해당 코드는 Python 코드 내부의 ORM(SQLAlchemy)에 대한 코드로

ORM은 테이블의 설계도를 메모리에 미리 등록하는 설계도 보관소 라고 생각하면 된다.

Python 스크립트가 실행될때, class ModelTest(SQLModel, table=True): 라는 코드를 읽어들이는 시점에 메모리에 ModelTest라는 이름의 설계도가 이미 등록되었는지를 확인한다.

ORM에 이미 같은 이름으로 등록된 설계도가 있다면 SQLAlchemy는 기본적으로 오류를 발생시키는데, 위 코드가 있다면 오류를 내지말고 기존 설계도를 이 새 코드로 갱신해줘 라고 명령하게 된다.

이 기능은 주로 서버의 reload=True 처럼 파이썬 모듈이 여러번 로드될때 메모리의 충돌을 방지하기 위해서 사용된다.

SQLModel.metadata.create_all(engine, checkfirst=True)

이 코드는 ORM이 아니라 실제 데이터베이스 서버에 테이블을 만들라는 요청 단계에서 기능을 하는 코드이다.

create_all() 함수가 실행되어 데이터베이스에 CREATE TABLE 명령을 보내기 직전에 실제 데이터베이스 내에 modeltest라는 명칭의 테이블이 존재하는지를 확인한다.

이 확인 범위는 MetaData 객체에 등록된 모든 정보를 기반으로 동작한다.

먼저 modeltest라는 이름의 테이블이 있는지 확인하고 테이블이 있다면 create table명령을 실행한다

그러고 modeltest라는 테이블 내에 ix_modeltest_name과 같은 인덱스가 존재하는지 확인하고 없다면 create index 명령을 실행한다

그 이후 제약 조건을 확인하고 생성하는 과정으로 진행된다.

11. 모델의 분리

마지막으로 모델을 사용 용도에 따라서 분리를 해볼 것 이다.

모델을 분리하는 이유는 보안(비밀번호, 내부 플래그 등은 DB에는 보여도 사용자에게 보여줄 응답에서는 필요 없음)과 요청/응답 규칙을 분리(Create에서는 필수인 항목이 Read나 Update에서는 필수적이지 않는 등의 필요한 규칙이 상황마다 다르기 때문)하고 DB 변경 내성(DB에 컬럼을 생성하는 등의 추가적인 변경이 있더라도 각 상황마다 컬럼을 필수와 비필수로 분리해두면 변경이 생겨도 에러가 발생할 가능성이 낮기 때문, 추후에 아예 파일을 분리해서 변경된 버전으로 사용할 수 있도록 변경하면 유지보수가 쉬워짐)에 이점이 있고 가독성/재사용(공통 필드는 Base에 두고 Create/Read/Update가 이를 상속하면 공통되는 부분을 쉽게 추가 제거할 수 있게 됨)이 가능하다는 장점이 있다.

기존에 SQLAlchemy를 사용할때는 ORM 모델과 Scheme를 위한 모델을 따로 만들어서 Scheme을 분리해서 사용했었는데 이는 ORM의 경우는 아래와 같이 DeclarativeBase라는 모델을 상속한 Base라는 모델을 상속 받아야만 Base.metadata에

class Base(DeclarativeBase):
    pass

테이블의 메타데이터를 등록하기 때문이다.

그렇기에 데이터 스키마를 위해서는 pydantic의 BaseModel을 사용해서 따로 선언을 해줘야 한다.

그런데 SQLModel을 상속하면 ORM 모델과 데이터 스키마 모델을 한번에 처리할 수 있다.

그렇기에 SQLModel을 사용할때는 모델을 분리할 때 ORM 모델을 포함해서 같이 분리가 가능하다.

가장 먼저 공통 부분이 될 ModelTestBase모델을 분리해보자

Base 모델

base 모델의 경우는 다수의 모델이 공통으로 사용할만한 필드를 분리해서 만들어준다.

모델의 형태가 위와 같을 경우 이름, 나이, 풀네임에 대해서는 외부에서 입력을 받아서 사용되며 테이블에 insert, update를 할때 사용된다.

id의 경우는 서버측에서 생성하거나 시퀀스 등으로 DB에서 생성해 같이 넣어주는게 보통이기 때문에 id의 경우는 외부에서 값을 받는 등의 작업이 진행되지 않는다.

그렇기에 공통으로 잡힐 데이터는 이름, 나이 , 풀네임으로 해당 부분에 대해서 Base 모델로 분리를 해주자.

이제 ModelTest에서는 겹치는 항목들에 대해서는 제거해주고

SQLModel 대신에 Base 모델을 상속해주면 된다.

먼저 ORM 모델을 분리했으니 데이터 모델(API 모델)을 분리해서 만들어보자.

먼저 insert에 사용될 create 모델을 만들어보자.

API 모델 : Create

create 모델은 DB에 insert 하기 위해서 사용되는 데이터베이스 스키마이자 API 모델이다.

해당 모델은 사용자에게 입력 받는 부분만을 넣어 생성한다.

그러면 사실상 Base 모델과 동일하기에 상속만 하고 내부 로직은 따로 추가하지 않으면 된다 .

API 모델 : Update

update모델의 경우는 DB에 변경할 수 있는 컬럼에 대해서만 필드를 선언하면 된다.

이때 optional로 데이터를 받지 않아도 되게 끔 만들어 줘야 선택적으로 원하는 필드만 데이터를 받아 원하는 컬럼만을 수정할 수 있도록 할 수 있다.

추가로 주의할 점은 기존의 Base모델을 보면

full_name은 str로 무조건 받아야 하는 형태로 구현되어 있는 것을 볼 수 있는데 update에서는 선택적으로 데이터를 받을 수 있어야 하기 때문에 해당 필드를 오버라이드 해줘야한다.

API 모델 : Select

select의 경우는 실제 데이터를 받아서 DB로 가는데에 따로 select 관련 모델을 사용하지는 않고 기존의 ORM모델을 그대로 사용한다.

대신에 사용자에게 응답을 만들때에는 response_model을 사용할때 select 모델이 사용되게 된다.

이는 ORM모델에서 응답에 반환하지 말아야하는 컬럼이 있다고 하더라도 select 모델에 필드로 설정되어 있지 않다면 포함시키지 않고 내보내주는 장점이 있다.

만들어보자면 우리가 갖고 있는 필드 중에서는 따로 보여주지말아야 할 부분은 없기 때문에 그냥 상속해서 사용하면 되는데 일단 id는 필수가 될 수 있도록 오버라이딩 해주고

만약에 ORM객체에

이런식으로 보여주면 안되는 필드를 갖고 있다고 한다면

Read를 따로 생성했을땐 해당 컬럼을 밖으로 반출하지 않게 된다.

그리고 Read의 model_config에 ConfigDict에 dict 말고도 속성을 읽어 올 수 있게 하는 설정인 from_attributes를 true로 설정해주면

모델이 속성에서도 값을 읽어 올수 있도록 설정할 수 있다.

API 모델 : Delete

추가로 Delete의 경우는 별도 모델 없이 보통 파라미터로 id(PK)만 받아서 삭제하는 방식을 사용한다.

@app.delete("/models/{item_id}", status_code=204)
def delete_model(item_id: str, session: Session = Depends(get_session)):
    obj = session.get(ModelTest, item_id)
    if not obj:
        # 팀 컨벤션에 따라 404 또는 204로 일관되게 처리
        raise HTTPException(404, "not found")
    session.delete(obj)
    session.commit()
    # 204라서 바디 없음

다만 다건 Delete의 경우는 id 값을 List로 받아서 사용할 수도 있는데 이때는 모델을 만들어서 사용하기도 한다.

class Ids(SQLModel):
    ids: List[str]

@app.post("/models/bulk-delete", status_code=204)
def bulk_delete(body: Ids, session: Session = Depends(get_session)):
    stmt = sa_delete(ModelTest).where(ModelTest.id.in_(body.ids))
    session.exec(stmt)
    session.commit()