[Backend] MMF 백엔드 개발 현황 – 2025 – 01 – 14 ( 1 )

사용자 관련

MMF를 위한 데이터베이스 새로 생성

( 현재 사용자 테이블 한 개 존재 )

주요 코드

models.py – 모델 선언

from sqlalchemy import Column, String
from database import Base

class User(Base):
    __tablename__ = "user"
    id = Column(String, primary_key = True)
    pw = Column(String, nullable = False)
    email = Column(String, nullable = False, unique = True)
    role = Column(String, nullable = False)

database.py – DB 연동

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

SQLALCHEMY_DATABASE_URL = "postgresql://postgres:3636@localhost/mmf_backend"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


def get_db():
    db: Session = SessionLocal()  # 세션 생성
    try:
        yield db  # 세션을 반환 (FastAPI는 이를 의존성 주입으로 사용)
    finally:
        db.close()

crud_user.py – CRUD ( Create, Read, Update, Delete ) 선언

from sqlalchemy.orm import Session
from models import User
import bcrypt

# 회원 가입
def register(db: Session, user_id: str, password: str, email: str, role: str):
    new_user = User(
        id=user_id,
        pw=password,
        email=email,
        role=role
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

# 회원 삭제
def delete_user(db: Session, email: str):
    user = db.query(User).filter(User.id == id).first()
    if user:
        db.delete(user)
        db.commit()
        return True
    return False

def update_user(db: Session, id: str, new_id : str, new_pw: str, new_email: str):
    try:
        user = db.query(User).filter(User.id == id).one()

        if new_id:
            user.id = new_id
        if new_pw:
            user.pw = new_pw
        if new_email:
            user.email = new_email
        db.commit()
        return {"message" : "User Updated Successfully."}
    except Exception as e:
        db.rollback()
        return {"error" : str(e)}


def authenticate_user(db: Session, email: str, password: str):
    # 이메일로 사용자 조회
    user = db.query(User).filter(User.email == email, User.pw == password).first()

    # 사용자가 없으면 None 반환
    if not user:
        return None

    return user

user_router.py – 사용자 관련된 기능에 한해서만 라우팅

import jsonify
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse
from database import SessionLocal, get_db
from models import User
from pydantic import BaseModel
from sqlalchemy.orm import Session
from crud_user import register, delete_user, update_user, authenticate_user

router = APIRouter()

class RegisterRequest(BaseModel):
    name: str
    password: str
    email: str

@router.post("/register")
def register_user(request: RegisterRequest, db: Session = Depends(get_db)):
    try:
        new_user = register(db = db, user_id = request.name, password=request.password, email=request.email, role="user")
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Error: {str(e)}")
    return {"message": "Register Success"}

class LoginRequest(BaseModel):
    email: str
    password: str

@router.post('/login')
def login(request: LoginRequest, db: Session = Depends(get_db)):
    user = authenticate_user(db, request.email, request.password)

    if user is None:
        print("user is None")
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # 로그인 성공 후 역할에 따라 다른 메시지 반환
    if user.role == "admin":
        return JSONResponse(content={"message": "관리자님 반갑습니다.", "role": "admin"}, status_code=200)
    elif user.role == "user":
        return JSONResponse(content={"message": f"{user.id} 반갑습니다.", "role": "user"}, status_code=200)
    else:
        raise HTTPException(status_code=401, detail="회원정보가 없습니다.")

main.py – fastAPI 앱 생성

from fastapi import FastAPI
from database import engine
from models import User
import crud_user
from user_router import router
from fastapi.middleware.cors import CORSMiddleware

# uvicorn main:app --host 0.0.0.0 --port 5000    (터미널로 실행시킬 때)

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 모든 도메인에서 접근 허용, 실제 운영 환경에서는 특정 도메인만 허용하는 것이 좋음
    allow_credentials=True,
    allow_methods=["*"],  # 모든 HTTP 메서드 허용
    allow_headers=["*"],  # 모든 헤더 허용
)


app.include_router(router)

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI on port 5000!"}