[웹서버] PostgreSQL 연결 + User CRUD

오규성·2026년 1월 15일
FastAPIserver

웹 서버 [FastAPI]

목록 보기
2/4

1편에서 FastAPI 기초 설정을 마쳤다. 이번 편에서는 PostgreSQL 을 연결하고, 계층형 아키텍처 패턴 구조로 User CRUD API를 만들어 보려고 한다.

구현할 전체 프로젝트 구조는 다음과 같다.

main.pyFastAPI 앱 진입점
api/HTTP 요청 처리
schemas/요청/응답 검증
crud/DB 조작 로직
models/DB 테이블 정의
core/설정, DB 연결

1. PostgreSQL 설치 및 실행

터미널에 다음을 입력하자

brew install postgresql # postgreSql 설치
brew services start postgresql # postgreSql 실행

이제 postgres 에 접속하여 테이블과 DB 를 생성할 것이다.

psql postgres # psql 을 통해 postgres 데이터베이스 관리 터미널 접속
CREATE USER {이름} WITH PASSWORD {'mypassword'}; # 유저 생성
CREATE DATABASE {DB 이름} OWNER {위에서 작성한 이름}; # postgres 에서 관리할 DB 생성
\q # 탈출

2. sqlalchemy 설치

가상환경에 접속한 상태로 sqlalchemy 를 설치해준다.
만약 접속 상태가 아니라면 다음을 입력하여 가상환경을 실행한다.

source .venv/bin/activate

이후 다음을 입력한다.

pip install sqlalchemy psycopg2-binary pydantic-settings alembic

위에서 설치하는 것들은 각각 다음과 같다.

sqlalchemy

  • Python ORM (Object-Relational Mapping, 객체 관계형 매핑) 으로 SQL 쿼리문을 문자열로 쓰지 않고, 파이썬 클래스 (Entity) 를 만들어 데이터를 관리할 수 있게 해준다.

psycopg2-binary

  • 파이썬이 PostgreSQL 데이터베이스와 통신할 수 있게 해주는 드라이버

pydantic-settings

  • DB 주소, 비밀번호 같은 민감한 설정 정보를 안전하게 관리해주는 라이브러리로, .env 파일에 저장된 환경 변수를 파이썬 객체로 읽어온다.

alembic

  • DB 마이그레이션 라이브러리

3. 환경설정

3-1. env 파일 설정

# 형식: postgresql+psycopg2://유저:비번@호스트:포트/DB이름
# localhost:5432, 포트 번호가 5432인 이유는 PostgreSQL 의 기본 포트이기 때문 !
DATABASE_URL=postgresql+psycopg2://your_username:your_password@localhost:5432/your_dbname

3-2. app/core/config.py 등록

Project - App - Core 폴더를 만들고, config.py 파일을 생성 후 다음 내용을 입력한다.

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
	# DATABASE_URL 은 필수여야하므로 타입 선언
    DATABASE_URL: str
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

settings = Settings()

이 코드를 통해 Pydantic 라이브러리는 .env 파일에서 DATABASE_URL 을 가져오고, settings = Settings() 를 통해 싱글톤 객체를 생성한다.

3-3. app/core/database.py 등록

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

# pool_pre_ping 은 연결 확인 기능임. 쿼리 전송 전 연결상태를 확인하여 미연결 시 재연결 시도
engine = create_engine(settings.DATABASE_URL, pool_pre_ping=True)
# autocommit = 데이터를 넣자마자 자동 저장 여부, False 시 db.commit() 호출해야 저장
# autoFlush = 쿼리 실행 전에 자동 flush를 할 지 여부
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

def get_db():
    """FastAPI에서 Depends(get_db)로 사용"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

4. 모델 정의하기

여기서 실제 DB 에서 사용할 User Class 를 정의할 것이다.

4-1. app/models/base.py

모든 모델이 기본적으로 상속해야하는 Base 클래스를 생성해준다.

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

4-2. app/models/user.py

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base

class User(Base):
    __tablename__ = "users"

	# Mapped 로 DB Column 과 연결 상태라는 것을 인식하게 하고 파이썬에서 취급할 자료형을 설정
    # mapped_column 으로 DB Table 에서 사용할 타입을 설정
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    name: Mapped[str | None] = mapped_column(String(10), nullable=True)

5. Schema 정의

이곳에서는 API 요청/응답용 클래스를 생성한다.

5-1. app/schemas/user.py 생성

from pydantic import BaseModel, ConfigDict

class UserCreate(BaseModel):
    email: str
    name: str | None = None

class UserUpdate(BaseModel):
    name: str | None = None

class UserRead(BaseModel):
    id: int
    email: str
    name: str | None

	# ORM 객체 (SQLAlchemy 모델) 을 Pydantic 모델로 변환해주는 속성
    # Pydantic 은 Dict 형태로만 읽을 수 있음 ex) user["email"]
    # SQLAlchemy 는 Attribute 접근 방식 ex) user.email
    # 이를 자동변환해주는 옵션임
    model_config = ConfigDict(from_attributes=True)

6. CRUD 구현

클라이언트에서 API 요청을 하는 경우 진행할 처리 코드를 구현한다.
이전에 sessionmaker 객체 생성 당시 autocommit 을 False 로 주었으므로 DB 에 수정사항이 생기는 경우 db.commit() 을 실행하여 데이터를 저장해주어야 한다.

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate

class UserRepository():
    def __init__(self, db: Session):
        self.db = db
    
    def get_user(self, user_id: int) -> User | None:
        return self.db.get(User, user_id)


    def get_user_by_email(self, email: str) -> User | None:
        stmt = select(User).where(User.email == email)
        # 1개만 가져오거나 가져오지 않음. 2개 이상이면 에러 발생
        return self.db.execute(stmt).scalar_one_or_none()


    def list_users(self, limit: int = 20, offset: int = 0) -> list[User]:
        stmt = select(User).offset(offset).limit(limit)
        return list(self.db.execute(stmt).scalars().all())


    def create_user(self, data: UserCreate) -> User:
        user = User(email=data.email, name=data.name)
        self.db.add(user)
        self.db.commit()        # 실제 DB 반영
        self.db.refresh(user)   # DB에서 자동 생성된 값(id 등)을 user 객체에 채움
        return user


    def update_user(self, user: User, data: UserUpdate) -> User:
        # PATCH니까 들어온 값만 반영
        if data.name is not None:
            user.name = data.name

        self.db.commit()
        self.db.refresh(user)
        return user


    def delete_user(self, user: User) -> None:
        self.db.delete(user)
        self.db.commit()

7. API 라우터 구현

라우터 구현 전 app.core.deps.py 를 생성하여 다음 코드를 생성해주자.
해당 파일은 의존성을 관리하는 파일로 사용할 것이다.

from fastapi import Depends
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.crud.user import UserRepository

def get_user_repository(db: Session = Depends(get_db)) -> UserRepository:
    return UserRepository(db)

다음으로 app.api.v1 의 users.py 를 생성하고 다음 코드를 넣어주자.

from fastapi import APIRouter, Depends, HTTPException

from app.schemas.user import UserCreate, UserRead, UserUpdate
from app.crud.user import UserRepository
from app.core.deps import get_user_repository

router = APIRouter(prefix="/users", tags=["users"])

    
@router.post("", response_model=UserRead)
def create_user(body: UserCreate, user_repo: UserRepository = Depends(get_user_repository)):
    if user_repo.get_user_by_email(body.email):
        raise HTTPException(status_code=409, detail="Email already exists")
    return user_repo.create_user(body)


@router.get("/{user_id}", response_model=UserRead)
def get_user(user_id: int, user_repo: UserRepository = Depends(get_user_repository)):
    user = user_repo.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User Not found")
    return user


@router.get("", response_model=list[UserRead])
def list_users(limit: int = 20, offset: int = 0, user_repo: UserRepository = Depends(get_user_repository)):
    return user_repo.list_users(limit=limit, offset=offset)


@router.patch("/{user_id}", response_model=UserRead)
def patch_user(user_id: int, body: UserUpdate, user_repo: UserRepository = Depends(get_user_repository)):
    user = user_repo.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User Not found")
    return user_repo.update_user(user, body)


@router.delete("/{user_id}")
def delete_user(user_id: int, user_repo: UserRepository = Depends(get_user_repository)):
    user = user_repo.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="UserNot found")
    user_repo.delete_user(user)
    return {"ok": True}

8. main.py 등록 및 실행

from fastapi import FastAPI

from app.api.v1.users import router as users_router

app = FastAPI()

app.include_router(users_router)

9. Alembic 설정

9-1. Alembic 초기화

VSC 의 터미널을 실행하여 다음 코드를 입력해준다.

alembic init alembic

위의 코드를 실행하면 다음과 같이 alembic 폴더가 생성된 것을 확인할 수 있다.

9-2. alembic - env.py 수정

from app.models.base import Base
from app.models import user as user_model
from app.core.config import settings

우선 위의 클래스들을 import 해준다.
env.py 내부를 살펴보면 config, target_metadata 가 존재할텐데, 이를 다음과 같이 수정해준다.

config = context.config
config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)

target_metadata = Base.metadata

모든 수정을 완료한 경우 터미널로 다시 이동하여 다음을 입력한다.

# 마이그레이션 파일 생성
alembic revision --autogenerate -m "create users table"
# DB에 적용
alembic upgrade head

10. 서버 실행, docs 확인 및 테스트

서버를 실행하여 테스트 해보자.

uvicorn app.main:app --reload

http://localhost:8000/users 에 접속하면 초기 상태는 user Table 에 아무런 데이터가 없으므로 아무것도 나오지 않는다.

http://localhost:8000/docs 에 접속해보자.

FastAPI 는 Swagger UI 를 자동구현해주어 등록된 Router 에 대한 정보들을 한눈에 확인 및 테스트 할 수 있게 해주는 기능이 존재한다.

이곳에서 POST -> Try It Out 을 클릭해주고, 전송할 Body 를 입력해준다.

실행 결과 200 이 나왔으므로 정상적으로 서버 통신이 진행되었다는 것을 알 수 있다.
이제 아까 http://localhost:8000/users 에 접속했던 브라우저를 새로고침 or 재접속해보자.

방금 등록한 유저 정보가 정상적으로 나오는 것을 확인할 수 있다.

주의 !

  • 만약 코드를 정상 입력하였음에도 에러가 발생하는경우 코드 저장을 하였는지 확인해보자.
  • 저장을 직접 하기가 귀찮다면 VSC - 파일의 자동 저장을 체크해주자.

아직 웹서버 구현 기초밖에 모르는 터라 동기 방식으로만 글을 쓰고 있어서 부족한 점이 많지만, 서버 부분을 익혀가며 클라이언트와 병행 공부하면 좋은 면이 많을 것 같다. 다음 장에는 JWT 관련 글을 작성하려고 하는데, 아마 배포까지 마무리하면 웹서버 관련은 아주 가끔만 올리지 않을까 싶다.

JWT 글을 썻었는데 문제가 많아 보여서 삭제했습니다 !

profile
오규성
안드로이드 개발자 Gyu 의 개발 블로그 !
이전 포스트

[웹서버] VSC + FastAPI 기초 설정

다음 포스트

[웹서버] Docker 찍먹해보기

0개의 댓글