[파이썬] FastAPI - 메모 앱 프로젝트 8: MVC 패턴

이전 내용

[파이썬] FastAPI - 메모 앱 프로젝트 7: 보완(회원가입 등)

---

MVC 패턴

 

FastAPI 애플리케이션에 MVC 패턴을 적용하는 것은 코드의 구조화와 모듈화를 통해 애플리케이션의 관리 및 유지보수를 향상시키는 데 매우 유용하다. MVC 패턴은 각각의 구성 요소를 분리하여 코드의 가독성을 높이고, 개발 과정에서의 관심사를 명확히 분리함으로써 개발 효율성을 증가시킨다.

MVC 패턴은 대규모 애플리케이션 또는 여러 개발자가 협업하는 프로젝트에서 특히 권장된다.

• M(모델): 애플리케이션의 데이터 나타냄. (데이터의 상태 변경, 저장, 업데이트 등 작업 수행)
• V(뷰): 텍스트, 체크박스 등의 사용자 인터페이스 나타냄. (사용자에게 보여지는 화면)
• C(컨트롤러): 데이터와 비즈니스 로직 사이의 상호동작 관리. (사용자의 입력을 받아 모델을 업데이트하고 변경된 데이터 뷰에 전달 )

MVC 패턴에 대해 궁금하면 하단 링크 참고

[웹개발] MVC(Model-View-Controller) 패턴이란?

---

MVC 패턴 적용 구조

 

여태까지 FastAPI 애플리케이션의 모든 코드를 main.py에 작성했는데, 각 부분의 책임을 명확화하고 유지보수성을 향상시키기 위해 main.py에 작성된 코드를 MVC 패턴에 따라 나눈다.

• 모델: 데이터와 데이터 처리 담당
• 뷰: 사용자 인터페이스와 관련된 로직 처리
• 컨트롤러: 사용자의 요청에 따라 모델과 뷰 사이의 상호 작용 관리

◆ 모델: models.py (데이터 모델)

• 데이터베이스의 테이블과 관계를 정의하는 모델 담당
• SQLAlchemy ORM을 사용하여 데이터베이스 스키마를 정의하고, 이 파일에서 정의된 클래스는 데이터베이스 테이블과 직접적으로 상응한다.

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from database import Base

# 사용자 모델 정의
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, index=True) # 정수형, PK
    username = Column(String(100), unique=True, index=True) # 중복 불가
    email = Column(String(200))
    hashed_password = Column(String(512))
    
# Memo 모델 정의
class Memo(Base):
    __tablename__ = 'memo'
    id = Column(Integer, primary_key=True, index=True) # 정수형, PK 설정
    user_id = Column(Integer, ForeignKey('users.id')) # 사용자 참조 추가
    title = Column(String(100), nullable=False) # 제목, 문자열, 값이 없는 예외 경우 방지
    content = Column(String(1000), nullable=False) # 내용, 문자열, 값이 없는 예외 경우 방지

    user = relationship("User") # 사용자와의 관계 설정

 

◆ 컨트롤러: schemas.py (데이터 검증 및 직렬화)

• 컨트롤러와 뷰의 일부 기능을 담당하며, Pydantic을 사용하여 데이터의 직렬화 및 유효성 검사를 수행하는 데 활용된다.
• 클라이언트로부터 받은 데이터의 유효성을 검사하고, API 응답 형식을 정의한다.

from pydantic import BaseModel
from typing import Optional

# 회원 가입 시 데이터 검증
class UserCreate(BaseModel):
    username: str
    email: str
    password : str # 해시 전 패스워드

# 회원 로그인 시 데이터 검증
class UserLogin(BaseModel):
    username: str
    password: str # 해시 전 패스워드

# pydantic 사용하여 데이터 검증 수행
class MemoCreate(BaseModel):
    title: str
    content: str

# BaseModel 상속받아 메모 수정 정의하는 클래스
class MemoUpdate(BaseModel):
    title: Optional[str] = None # 문자열 타입, None 값 가능
    content: Optional[str] = None # 문자열 타입, None 값 가능

 

◆ 컨트롤러: controllers.py (경로 핸들링 및 뷰 로직)

• 사용자의 요청을 처리하고 적절한 응답을 반환하는 컨트롤러
• FastAPI의 APIRouter를 사용하여 각 경로에 대한 요청 처리 로직 구현. 데이터를 받아 처리하고, 결과 반환.
• FastAPI의 APIRouter: 여러 API 엔드포인트를 그룹화하여 구성할 수 있도록 도와주는 기능. 이를 통해 코드 구조를 더 모듈화하고 관리하기 쉽게 만들어 주며, 특정 경로에 관련된 경로를 하나의 라우터에서 정의할 수 있다.
• FastAPI의 APIRouter 사용을 위해서는 router = APIRouter()로 라우터를 생성하고, 다양한 경로 및 메서드를 추가한 후, 최종적으로 FastAPI 애플리케이션에 포함시킨다. (@app.HTTP 메서드 → @router.HTTP 메서드로 변경)

from fastapi import FastAPI, Request, Depends, HTTPException, APIRouter
from sqlalchemy.orm import Session
from models import User, Memo # 모델 import
from schemas import UserCreate, UserLogin, MemoCreate, MemoUpdate # 스키마 import
from dependencies import get_db, get_password_hash, verify_password # 의존성 import
from fastapi.templating import Jinja2Templates
import re

router = APIRouter()
templates = Jinja2Templates(directory="templates")

# 회원 가입
@router.post("/signup")
async def signup(signup_data: UserCreate, db: Session=Depends(get_db)):
    # ID 규칙 확인: 영어 소문자와 숫자만 허용
    if not re.fullmatch(r"[a-z0-9]+", signup_data.username):
        raise HTTPException(status_code=400, detail="사용자 이름은 영어 소문자와 숫자로만 구성되어야 합니다.")

    # 비밀번호 규칙 확인
    password_regex = re.compile(r"""
        (?=.*[a-z])          # 적어도 하나의 소문자
        (?=.*[A-Z])          # 적어도 하나의 대문자
        (?=.*\d)             # 적어도 하나의 숫자
        (?=.*[@$!%*?&\-_+=<>]) # 적어도 하나의 특수문자
        .{10,}               # 길이는 10자 이상
    """, re.VERBOSE)

    if not password_regex.match(signup_data.password):
        raise HTTPException(status_code=400, detail="비밀번호는 최소 10자 이상이며, 대문자, 소문자, 숫자 및 특수문자가 포함되어야 합니다.")
    # username 중복 확인
    existing_user = db.query(User).filter(User.username == signup_data.username).first()
    if existing_user:
        raise HTTPException(status_code=400, detail="이미 존재하는 사용자 이름 입니다.")
    hashed_password = get_password_hash(signup_data.password) # 비밀번호 해시 기능
    new_user = User(username=signup_data.username, email=signup_data.email, hashed_password=hashed_password)
    db.add(new_user)
    
    try:
        db.commit()
    except Exception as e:
        db.rollback() # 에러 발생 시 롤백
        raise HTTPException(status_code=500, detail="회원 가입 실패. 다시 시도해 주세요.")
    db.refresh(new_user)
    return {"message": "회원가입을 성공하였습니다."}

# 로그인
@router.post("/login")
async def login(request: Request, signin_data: UserLogin, db: Session=Depends(get_db)):
    user = db.query(User).filter(User.username == signin_data.username).first()
    if user and verify_password(signin_data.password, user.hashed_password):
        request.session["username"] = user.username
        return {"message": "로그인 성공!"}
    else:
        raise HTTPException(status_code=401, detail="로그인이 실패하였습니다.")

# 로그아웃
@router.post("/logout")
async def logout(request: Request):
    request.session.pop("username", None)
    return {"message": "로그아웃 완료!"}


# 메모 생성
@router.post("/memos")
async def create_user(request: Request, memo: MemoCreate, db:Session = Depends(get_db)):
    username = request.session.get("username")
    if username is None:
        raise HTTPException(status_code=401, detail="Not Authorized")
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User를 찾을 수 없습니다.")
    new_memo = Memo(user_id=user.id, title = memo.title, content = memo.content)
    db.add(new_memo)
    db.commit()
    db.refresh(new_memo)
    # 새로 생성된 사용자 정보 반환
    return new_memo

# 메모 조회
@router.get("/memos")
async def list_memos(request: Request, db: Session = Depends(get_db)):
    # 사용자별 메모 관리
    username = request.session.get("username")
    if username is None:
        raise HTTPException(status_code=401, detail="Not Authorized")
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User를 찾을 수 없습니다.")
    memos = db.query(Memo).filter(Memo.user_id == user.id).all()
    # 리스트 내포 사용하여 전체 정보 반환
    return templates.TemplateResponse("memos.html", {
        "request": request, 
        "memos": memos,
        "username": username}) # 사용자 이름 추가

# 메모 수정
@router.put("/memos/{memo_id}")
async def update_user(request:Request, memo_id: int, memo: MemoUpdate, db: Session = Depends(get_db)):
    username = request.session.get("username")
    if username is None:
        raise HTTPException(status_code=401, detail="Not Authorized")
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User를 찾을 수 없습니다.")
    db_memo = db.query(Memo).filter(Memo.id == memo_id).first()
    if db_memo is None:
        return {"error": "User를 찾을 수 없습니다."}
    
    # 사용자 정보 업데이트
    if memo.title is not None:
        db_memo.title = memo.title
    if memo.content is not None:
        db_memo.content = memo.content

    db.commit()
    db.refresh(db_memo)
    return db_memo

# 메모 삭제
@router.delete("/memos/{memo_id}")
async def delete_user(request:Request, memo_id: int, db: Session = Depends(get_db)):
    username = request.session.get("username")
    if username is None:
        raise HTTPException(status_code=401, detail="Not Authorized")
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User를 찾을 수 없습니다.")
    db_memo = db.query(Memo).filter(Memo.id == memo_id, Memo.user_id == user.id).first()
    if db_memo is None:
        return {"error": "Memo를 찾을 수 없습니다."}
    db.delete(db_memo)
    db.commit()
    return {"message": "Memo가 삭제되었습니다."}

# 라우트
@router.get('/')
async def read_root(request: Request):
    return templates.TemplateResponse('home.html', {"request": request})

# 'About' 페이지 추가
@router.get('/about')
async def about():
    return {"message": "메모 앱 소개 페이지 입니다."}

 

◆ 컨트롤러: dependencies.py (의존성 관리)

• 컨트롤러의 일부 기능을 담당하며, 의존성 주입과 관련된 로직을 포함한다.
• 데이터베이스 세션 관리 및 사용자 인증과 관련된 유틸리티 함수 포함.

from sqlalchemy.orm import sessionmaker, Session, relationship
from passlib.context import CryptContext
from database import SessionLocal

# passlib을 사용한 사용자 인증
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

• get_password_hash(password: str) -> str: pwd_context.hash(password) 메소드를 사용하여 주어진 비밀번호를 해시하고, 그 결과를 문자열 형태로 반환합니다. 이렇게 해시된 비밀번호는 데이터베이스에 안전하게 저장될 수 있다.
• verify_password(plain_password: str, hashed_password: str) -> bool: pwd_context.verify(plain_password, hashed_password) 메소드를 사용하여 원본 비밀번호(plain_password)와 해시된 비밀번호(hashed_password)를 비교합니다. 일치할 경우 True를 반환하고, 그렇지 않으면 False를 반환

◆ 모델: database.py (데이터베이스 설정)

• 모델의 일부 기능 담당. 데이터베이스 연결 및 설정 관리
• 데이터베이스 엔진을 설정하고, SQLAlchemy의 sessionmaker를 통해 데이터베이스 세션 생성

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

# 데이터베이스 URL 설정
# postgresql://user명:비밀번호@host이름/db이름
DATABASE_URL = "postgresql://postgres:1234@localhost/memo_app" # 실무에서는 노출되지 않는 것이 좋음

# 엔진 생성성
engine = create_engine(DATABASE_URL)

# 세션 생성성
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base 클래스 생성
Base = declarative_base()

 

◆ 뷰: templates (디렉터리)

• 사용자에게 보이는 부분 담당
• HTML 템플릿 파일들을 포함하여, Jinja2 템플릿 엔진을 통해 렌더링되는 동적인 웹페이지 구성

[main.py]

main.py 파일은 FastAPI의 진입점으로, 애플리케이션 인스턴스를 생성하고 라우터를 포함시키며, 필요한 미들웨어를 설정한다.

from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from controllers import router # 컨트롤러 라우터 import
from database import Base, engine # 데이터베이스 설정 import
from starlette.middleware.sessions import SessionMiddleware

# FastAPI 애플리케이션 생성
app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="your-secret-key")
templates = Jinja2Templates(directory="templates")

# 템플릿 디렉터리 설정
templates = Jinja2Templates(directory="templates")

# 라우터 포함
app.include_router(router)

# 데이터베이스 테이블 생성성
Base.metadata.create_all(bind=engine)

@router.get('/')
async def read_root(request: Request):
    return templates.TemplateResponse('home.html', {"request": request})

---

테스트 하기

 

main.py를 실행하여 서버를 실행하고 메모 앱이 제대로 작동되는지 확인해 본다.

 

---

[참고]

가장 빠른 풀스택을 위한 플라스크 & FastAPI

---

다음 내용