LLM 기반 추천 시스템 백엔드

미들웨어

 

우선 CORS를 통해서 프론트엔드와의 연결을 허용한다.

from fastapi import FastAPI

# FastAPI 시작  
app = FastAPI()

origins = [
    "http://localhost:5173",  # Vite 개발 서버
    "http://127.0.0.1:5173",  # 다른 로컬 주소
    "https://your-production-frontend.com"  # 배포 환경에서는 실제 도메인
]


app.add_middleware(
    CORSMiddleware,
    #allow_origins=["*"],  # 모든 도메인 허용
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],  # 모든 메서드 허용
    allow_headers=["*"],  # 모든 헤더 허용
)

 

 

DB 모델 정의

 

ORM 형식에 맞게 DB 모델을 정의한다. 

from sqlalchemy import ForeignKey
from sqlalchemy.orm import (
    DeclarativeBase,
    mapped_column,
    Mapped
)

class Base(DeclarativeBase):
    pass

class Users(Base):
    __tablename__ = "users"

    userId: Mapped[int] = mapped_column(primary_key=True)
    gender: Mapped[str]
    age: Mapped[int]
    occupation: Mapped[str]
    zipCode: Mapped[str]

class Movies(Base):
    __tablename__ = "movies"

    movieId: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    genre: Mapped[str]

class Ratings(Base):
    __tablename__ = "ratings"

    userId: Mapped[int] = mapped_column(ForeignKey("users.userId"), primary_key=True)
    movieId: Mapped[int] = mapped_column(ForeignKey("movies.movieId"), primary_key=True)
    rating: Mapped[float]
    timestamp: Mapped[int]


class Recommendations(Base):
    __tablename__ = "recommendations"

    userId: Mapped[int] = mapped_column(ForeignKey("users.userId"), primary_key=True)
    movieId: Mapped[int] = mapped_column(ForeignKey("movies.movieId"), primary_key=True)
    meanRating: Mapped[float]
    timestamp: Mapped[str]
    recommenderId: Mapped[int] = mapped_column(ForeignKey("recommenders.id"))
    recommenderName: Mapped[str] = mapped_column(ForeignKey("recommenders.model_name"))
    feedback: Mapped[str] = mapped_column(nullable=True)  # feedback은 선택 사항이므로 nullable=True 설정

 

 

비동기식 데이터베이스 선언

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import MetaData, create_engine
from sqlalchemy.orm import sessionmaker
import logging
import os


'''
databases는 비동기를 지원하는 라이브러리다.  
async와 await가 가능하다.  
반면에 database는 동기식이다. 
'''
# 상대 경로로 변경
DATABASE_URL = "sqlite+aiosqlite:///./ott.db"  # 현재 디렉토리에 생성되도록 수정

# 또는 절대 경로 사용
import os
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
print(f"backend base_dir: {BASE_DIR}")
DATABASE_URL = f"sqlite+aiosqlite:///{os.path.join(BASE_DIR, 'ott.db')}"

# MetaData 객체 생성  
metadata = MetaData()

database = 1

'''
# 동기 엔진
engine = create_engine(DATABASE_URL, echo=True)

# 동기 세션
Session = sessionmaker(bind=engine, expire_on_commit=False)
'''

# 비동기 엔진 생성
engine = create_async_engine(DATABASE_URL, echo=True)  # aiosqlite 사용

 

 

 

 

Search를 API의 GET 으로 구현

 

@app.get("/api/search", response_model=List[dict])
async def search_movies(query: str, db: AsyncSession = Depends(get_async_db)):
    if not query:
        raise HTTPException(status_code=400, detail="검색어를 입력해주세요")
    
    try:
        # ILIKE를 사용하여 대소문자 구분 없이 검색
        search_term = f"%{query}%"
        query = select(Movies).select_from(Movies).where(
            Movies.title.ilike(search_term) |  # 제목 검색
            Movies.genre.ilike(search_term)    # 장르 검색
        )
        
        results = await db.execute(query)
        movies = results.scalars().all()
        print(f"movies: {movies}")
        # Record 객체를 딕셔너리로 변환
        results_as_dict = []
        for movie in movies:
            movie_dict = {
                "movieId": movie.movieId,
                "title": movie.title,
                "genre": movie.genre
            }
            results_as_dict.append(movie_dict)
        
        if not results_as_dict:
            return [] # 없으면 빈칸 리턴 
            
        return results_as_dict
        
    except Exception as e:
        print(f"Search error: {str(e)}")  # 서버 로그에 에러 출력
        raise HTTPException(status_code=500, detail="검색 중 오류가 발생했습니다")

 

GET을 활용해서 주어진 쿼리에 해당하는 영화를 검색한다.

SELECT로 DB에서 해당되는 내역을 찾아서 반환한다.

대소문자를 모두 포함하기 위해서 ILIKE를 사용한다.

 

 

Rating 추가를 POST로 구현

 

@app.post("/api/ratings")
async def create_rating(rating: RatingBase, db: AsyncSession = Depends(get_async_db)):
    '''
    if not rating:
        raise HTTPException(status_code=404, detail='No rating object')
    else:
        raise HTTPException(status_code=200, detail='Right object')
    '''
    try:
        query = insert(Ratings).values(
            userId=rating.userId,
            movieId=rating.movieId,
            rating=rating.rating,
            timestamp=rating.timestamp
        )
        await db.execute(query)
        await db.commit()
        return {"message": "Rating added successfully"}
    except Exception as e:
        await db.rollback()
        raise HTTPException(status_code=500, detail=str(e))

 

DB에 유저가 영화에 새로 별점을 매겼을 때 이를 INSERT로 삽입한다.