오늘 작업
전체적인 코드 구조 변경, 최적화
로그인 구현

코드 구조 변경, 최적화

FastAPI를 공부하다가 자주쓰는 폴더 구조를 알게되어서 그에 맞게 코드를 전부 수정했다.

폴더 구조

project_name/
│
├── app/                    # 애플리케이션 코드를 포함하는 메인 폴더
│   │
│   ├── api/                # 엔드포인트와 라우터를 정의하는 폴더
│   │   ├── __init__.py
│   │   ├── dependencies.py # 의존성(예: get_db)을 정의하는 파일
│   │   └── api_v1/         # 버전 1 API 엔드포인트를 모아두는 폴더
│   │       ├── __init__.py
│   │       ├── endpoints/  # 엔드포인트별로 파일을 나누어 관리
│   │       │   ├── __init__.py
│   │       │   ├── user.py # 사용자 관련 엔드포인트 (회원가입, 로그인 등)
│   │       │   └── ...
│   │       └── router.py   # API v1의 라우터를 정의하고 하위 엔드포인트를 포함시키는 파일
│   │
│   ├── core/               # 설정과 보안 등의 코어 기능을 담는 폴더
│   │   ├── __init__.py
│   │   ├── config.py       # 프로젝트 설정(환경 변수 등)을 관리하는 파일
│   │   └── security.py     # 보안 관련 유틸리티(비밀번호 해싱, 토큰 생성 등)
│   │
│   ├── crud/               # 데이터베이스 CRUD 연산을 정의하는 폴더
│   │   ├── __init__.py
│   │   └── crud_user.py    # 사용자 데이터에 대한 CRUD 연산
│   │
│   ├── models/             # 데이터베이스 모델(Schema)을 정의하는 폴더
│   │   ├── __init__.py
│   │   └── user.py         # 사용자 모델 정의
│   │
│   ├── schemas/            # Pydantic 모델(데이터 검증 및 스키마)을 정의하는 폴더
│   │   ├── __init__.py
│   │   └── user.py         # 사용자 관련 스키마 (예: UserCreate, UserOut 등)
│   │
│   ├── database.py         # 데이터베이스 세션 관리 및 연결 설정
│   └── main.py             # FastAPI 애플리케이션 인스턴스와 미들웨어 설정
│
├── tests/                  # 테스트 코드
│   ├── __init__.py
│   └── test_api/
│       ├── __init__.py
│       └── test_user.py
│
├── requirements.txt        # 프로젝트 의존성 목록
└── .env                    # 환경 변수 설정 파일

이게 이상적인 프로젝트 구조라고한다. 그리닷 프로젝트는 앱에 많은 기능을 넣을 필요가 없기 때문에 참고할 것만 참고했다.

그리닷 폴더 구조

회원가입, 로그인만 구현된 우리의 폴더 구조는 이렇게 되어있다.

각각 어떤 생각을 가지고 구현했는지 중요하다고 생각한다.

리마인드해보면서 정리를 해놓는게 좋을 것 같다.

models 폴더

1. init.py

from sqlalchemy import create_engine
from app.models.base import Base
from app.models.models import Member, Gree, Logs
from app.core.config import DATABASE_URI

def init_db():
    engine = create_engine(DATABASE_URI)
    # Base.metadata.create_all을 호출하여 모든 상속된 테이블을 생성합니다.
    Base.metadata.create_all(engine)
    print("모든 테이블이 생성되었습니다.")

if __name__ == "__main__":
    init_db()

동기식으로 테이블을 다룰 경우는 초기에 테이블 생성할 경우밖에 없어서 따로 init에 넣어놨다.

2. base.py

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

SQLAlchemy 모델을 정의하기 위한 기반 클래스이며, 데이터베이스 테이블과 모델 클래스 간의 연결을 설정하는 역할을 한다.

굳이 뺄 필요는 없는 것 같지만 폴더 구조도를 참고해서 따로 빼놨다.

3. enums.py

from enum import unique, Enum as PyEnum


@unique
class RoleEnum(PyEnum):
    ADMIN = "ADMIN"
    MANAGER = "MANAGER"
    MEMBER = "MEMBER"


@unique
class StatusEnum(PyEnum):
    ACTIVATE = "ACTIVATE"
    DISABLED = "DISABLED"


@unique
class GradeEnum(PyEnum):
    FREE = "FREE"
    BASIC = "BASIC"
    PREMIUM = "PREMIUM"


@unique
class LogTypeEnum(PyEnum):
    SENDTALK = "SENDTALK"
    RECEIVEDTALK = "RECEIVEDTALK"
    CLICKED = "CLICKED"

테이블의 Enum 타입을 모아놓은 파일이다.

4. models.py

from sqlalchemy import Column, Integer, String, DATETIME, Enum, ForeignKey
from sqlalchemy.orm import relationship
from app.models.enums import RoleEnum, StatusEnum, GradeEnum, LogTypeEnum
from .base import Base

class Member(Base):
    __tablename__ = 'member'

    id = Column('member_id', Integer, primary_key=True, autoincrement=True)
    email = Column(String(255), nullable=False)
    nickname = Column(String(255), nullable=False)
    password = Column(String(255), nullable=False)
    role = Column(Enum(RoleEnum), nullable=False)
    status = Column(Enum(StatusEnum), nullable=False)
    grade = Column(Enum(GradeEnum), nullable=False)
    register_at = Column(DATETIME, nullable=False)

    gree = relationship("Gree", back_populates="member")
    logs = relationship("Logs", back_populates="member")

class Gree(Base):
    __tablename__ = 'gree'

    id = Column('gree_id', Integer, primary_key=True, autoincrement=True)
    member_id = Column(Integer, ForeignKey('member.member_id'), nullable=False)
    gree_name = Column(String(255), nullable=False)
    root_path = Column(String(255), nullable=False)
    prompt_character = Column(String(255))  # 대체될 수 있습니다
    prompt_age = Column(Integer)
    prompt_mbti = Column(String(255))  # 대체될 수 있습니다
    register_at = Column(DATETIME, nullable=False)

    member = relationship("Member", back_populates="gree")
    logs = relationship("Logs", back_populates="gree")

class Logs(Base):
    __tablename__ = 'logs'

    id = Column('log_id', Integer, primary_key=True, autoincrement=True)
    gree_id = Column(Integer, ForeignKey('gree.gree_id'), nullable=False)
    member_id = Column(Integer, ForeignKey('member.member_id'), nullable=False)
    log_type = Column(Enum(LogTypeEnum), nullable=False)
    talk = Column(String(255))
    register_at = Column(DATETIME, nullable=False)

    member = relationship("Member", back_populates="logs")
    gree = relationship("Gree", back_populates="logs")

SQLAlchemy를 사용하여 데이터베이스 모델을 정의하는 부분이다.

base.py에서 만든 Base를 가져와서 데이터베이스와 연결한다.

core 폴더

1. config.py

from pydantic.v1 import BaseSettings


class Settings(BaseSettings):
    API_v1_STR: str = "/api/v1"
    SECRET_KEY: str = "#나중에 정의#"
    ALGORITHM: str = "#나중에 정의#"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

settings = Settings()

# 데이터베이스 설정
DATABASE_URI = "mysql+pymysql://admin:비밀번호@database-1.c3mqckcawht2.ap-southeast-2.rds.amazonaws.com/greedot"
ASYNC_DATABASE_URI = "mysql+aiomysql://admin:비밀번호0@database-1.c3mqckcawht2.ap-southeast-2.rds.amazonaws.com/greedot"

이 설정과 데이터베이스 연결 정보는 FastAPI 애플리케이션의 설정 파일로 사용되며, 애플리케이션의 동작 및 데이터베이스 연결에 필요한 정보를 제공한다.

2. security.py

from datetime import datetime, timedelta
import jwt
from fastapi.security import OAuth2PasswordBearer
import bcrypt
from fastapi import HTTPException, status
from app.core.config import settings

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def hash_password(password: str) -> bytes:
    return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())


def verify_password(plain_password: str, hashed_password: str) -> bool:
    if isinstance(hashed_password, str):
        hashed_password = hashed_password.encode('utf-8')

    return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password)


def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

# 토큰을 검증하고 페이로드 반환, 유효하지 않은 경우 예외 발생
def verify_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM], options={"verify_exp": True})
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired")
    except jwt.PyJWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials")

비밀번호 해싱 및 사용자 토큰 생성, 검증 코드이다.

api_v1 폴더

1. endpoints 폴더 -> user.py

from fastapi import APIRouter, HTTPException, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from fastapi.security import OAuth2PasswordRequestForm
from datetime import datetime
from typing import Optional
from app.models.models import Member, RoleEnum, StatusEnum, GradeEnum
from app.core.security import hash_password, verify_password, create_access_token
from app.database import get_db
from pydantic import BaseModel

router = APIRouter()


# Pydantic 모델 정의
class RegisterRequest(BaseModel):
    email: str
    nickname: str
    password: str


class Token(BaseModel):
    access_token: str
    token_type: str

# 이메일 중복 확인
async def user_exists(email: str, db_session: AsyncSession) -> bool:
    async with db_session as session:
        result = await session.execute(select(Member).where(Member.email == email))
        member = result.scalars().first()
        return member is not None


## 이메일이 일치한지 확인
async def authenticate_user(email: str, password: str, db: AsyncSession) -> Optional[Member]:
    async with db as session:
        result = await session.execute(select(Member).where(Member.email == email))
        user = result.scalars().first()
        if user and verify_password(password, user.password):
            return user
    return None


@router.post('/register')
async def register_member(request: RegisterRequest, db: AsyncSession = Depends(get_db)):
    if await user_exists(request.email, db):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="이미 존재하는 사용자입니다.")
    if not request.email or not request.nickname or not request.password:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="모든 필드를 입력해주세요.")

    hashed_pwd = hash_password(request.password)

    new_member = Member(
        email=request.email,
        nickname=request.nickname,
        password=hashed_pwd,
        role=RoleEnum.MEMBER,
        status=StatusEnum.ACTIVATE,
        grade=GradeEnum.FREE,
        register_at=datetime.now()
    )
    db.add(new_member)
    await db.commit()
    await db.refresh(new_member)
    # 회원가입 후에 로그인 처리
    user = new_member
    return {"message": "회원가입이 성공적으로 완료되었습니다.", "user_id": user.id}


##일치하면 토큰 반환, 불일치하면 에러 메세지 반환
@router.post('/login', response_model=Token)
async def login_for_access_token(db: AsyncSession = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()):
    user = await authenticate_user(form_data.username, form_data.password, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.email})
    return {
        "access_token": access_token,
        "token_type": "bearer",
    }

실질적으로 user에 대한 기능이 들어있다.(회원가입, 로그인)

회원가입: member_id가 원래 존재하는지 비동기로 세션 생성 후 중복 확인, 중복이면 에러 메세지 반환 중복이 아니면 회원가입 성공 메세지 반환

로그인: 비동기 세션 생성 후 member_id 비교, 일치하면 토큰 반환 불일치하면 에러 메세지 반환

2. router.py

from fastapi import APIRouter
from app.api.api_v1.endpoints import user

api_router = APIRouter()
api_router.include_router(user.router, prefix="/user", tags=["user"])

APIRouter를 사용하여 API 엔드포인트들을 묶는 역할을 한다.

API가 크고 복잡한 경우, 코드를 모듈화하고 관리하기 위해 여러 개의 라우터를 사용하는 것이 좋다고 한다.

database.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.core.config import ASYNC_DATABASE_URI

# 비동기 엔진 생성
async_engine = create_async_engine(ASYNC_DATABASE_URI)

# 비동기 세션 생성자
AsyncSessionLocal = sessionmaker(bind=async_engine, class_=AsyncSession, expire_on_commit=False)


# 비동기 데이터베이스 세션 제공자 정의
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

비동기로 세션 생성 및 제공할때 이 코드를 불러서 사용한다.

main.py

import uvicorn
from fastapi import FastAPI
from app.api.router import api_router
from app.core.config import settings
app = FastAPI()


# API 라우터를 앱에 포함
app.include_router(api_router, prefix=settings.API_v1_STR)


if __name__ == "__main__":
    uvicorn.run(app, host="localhost", port=8000)

localhost 8080포트로 실행시킨다.

TIL

FastAPI의 폴더 구조를 알게 됐고 app.post 구현하던 회원가입 기능을
router.post로 구현하면서 모듈화의 차이를 알게 됐다.

app.post은 애플리케이션 레벨에서 전체 애플리케이션을 구성하는 데 사용되는 반면, router.post는 라우팅 모듈 내에서 특정 경로에 대한 핸들러를 정의하고 모듈을 재사용할 때 유용하게 사용된다.

이 프로젝트를 하면서 구현에 목매지 않고 좋은 코드가 뭔지 계속 생각하면서 성공하는 게 목표이다.