o2o-castad-backend/docs/analysis/orm_report.md

ORM 동기식 전환 보고서

개요

현재 프로젝트는 SQLAlchemy 2.0+ 비동기 방식으로 구현되어 있습니다. 이 보고서는 동기식으로 전환할 경우 필요한 코드 수정 사항을 정리합니다.

---

1. 현재 비동기 구현 현황

1.1 사용 중인 라이브러리

• sqlalchemy[asyncio]>=2.0.45
• asyncmy>=0.2.10 (MySQL 비동기 드라이버)
• aiomysql>=0.3.2

1.2 주요 비동기 컴포넌트

컴포넌트	현재 (비동기)	변경 후 (동기)
엔진	create_async_engine	create_engine
세션 팩토리	async_sessionmaker	sessionmaker
세션 클래스	AsyncSession	Session
DB 드라이버	mysql+asyncmy	mysql+pymysql

---

2. 파일별 수정 사항

2.1 pyproject.toml - 의존성 변경

파일: pyproject.toml

dependencies = [
    "fastapi[standard]>=0.115.8",
-   "sqlalchemy[asyncio]>=2.0.45",
+   "sqlalchemy>=2.0.45",
-   "asyncmy>=0.2.10",
-   "aiomysql>=0.3.2",
+   "pymysql>=1.1.0",
    ...
]

---

2.2 config.py - 데이터베이스 URL 변경

파일: config.py (라인 74-96)

class DatabaseSettings(BaseSettings):
    @property
    def MYSQL_URL(self) -> str:
-       return f"mysql+asyncmy://{self.MYSQL_USER}:{self.MYSQL_PASSWORD}@{self.MYSQL_HOST}:{self.MYSQL_PORT}/{self.MYSQL_DB}"
+       return f"mysql+pymysql://{self.MYSQL_USER}:{self.MYSQL_PASSWORD}@{self.MYSQL_HOST}:{self.MYSQL_PORT}/{self.MYSQL_DB}"

---

2.3 app/database/session.py - 세션 설정 전면 수정

파일: app/database/session.py

현재 코드 (비동기)

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from typing import AsyncGenerator

engine = create_async_engine(
    url=db_settings.MYSQL_URL,
    echo=False,
    pool_size=10,
    max_overflow=10,
    pool_timeout=5,
    pool_recycle=3600,
    pool_pre_ping=True,
    pool_reset_on_return="rollback",
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
)

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        except Exception as e:
            await session.rollback()
            raise e

변경 후 코드 (동기)

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from typing import Generator

engine = create_engine(
    url=db_settings.MYSQL_URL,
    echo=False,
    pool_size=10,
    max_overflow=10,
    pool_timeout=5,
    pool_recycle=3600,
    pool_pre_ping=True,
    pool_reset_on_return="rollback",
)

SessionLocal = sessionmaker(
    bind=engine,
    class_=Session,
    expire_on_commit=False,
    autoflush=False,
)

def get_session() -> Generator[Session, None, None]:
    with SessionLocal() as session:
        try:
            yield session
        except Exception as e:
            session.rollback()
            raise e

get_worker_session 함수 변경

- from contextlib import asynccontextmanager
+ from contextlib import contextmanager

- @asynccontextmanager
- async def get_worker_session() -> AsyncGenerator[AsyncSession, None]:
-     worker_engine = create_async_engine(
+ @contextmanager
+ def get_worker_session() -> Generator[Session, None, None]:
+     worker_engine = create_engine(
          url=db_settings.MYSQL_URL,
          poolclass=NullPool,
      )
-     session_factory = async_sessionmaker(
-         bind=worker_engine,
-         class_=AsyncSession,
+     session_factory = sessionmaker(
+         bind=worker_engine,
+         class_=Session,
          expire_on_commit=False,
          autoflush=False,
      )

-     async with session_factory() as session:
+     with session_factory() as session:
          try:
              yield session
          finally:
-             await session.close()
-     await worker_engine.dispose()
+             session.close()
+     worker_engine.dispose()

---

2.4 app/*/dependencies.py - 타입 힌트 변경

파일: app/song/dependencies.py, app/lyric/dependencies.py, app/video/dependencies.py

- from sqlalchemy.ext.asyncio import AsyncSession
+ from sqlalchemy.orm import Session

- SessionDep = Annotated[AsyncSession, Depends(get_session)]
+ SessionDep = Annotated[Session, Depends(get_session)]

---

2.5 라우터 파일들 - async/await 제거

영향받는 파일:

• app/home/api/routers/v1/home.py
• app/lyric/api/routers/v1/lyric.py
• app/song/api/routers/v1/song.py
• app/video/api/routers/v1/video.py

예시: lyric.py (라인 70-90)

- async def get_lyric_by_task_id(
+ def get_lyric_by_task_id(
      task_id: str,
-     session: AsyncSession = Depends(get_session),
+     session: Session = Depends(get_session),
  ):
-     result = await session.execute(select(Lyric).where(Lyric.task_id == task_id))
+     result = session.execute(select(Lyric).where(Lyric.task_id == task_id))
      lyric = result.scalar_one_or_none()
      ...

예시: CRUD 작업 (라인 218-260)

- async def create_project(
+ def create_project(
      request_body: ProjectCreateRequest,
-     session: AsyncSession = Depends(get_session),
+     session: Session = Depends(get_session),
  ):
      project = Project(
          store_name=request_body.customer_name,
          region=request_body.region,
          task_id=task_id,
      )
      session.add(project)
-     await session.commit()
-     await session.refresh(project)
+     session.commit()
+     session.refresh(project)
      return project

예시: 플러시 작업 (home.py 라인 340-350)

  session.add(image)
- await session.flush()
+ session.flush()
  result = image.id

---

2.6 서비스 파일들 - Raw SQL 쿼리 변경

영향받는 파일:

• app/lyric/services/lyrics.py
• app/song/services/song.py
• app/video/services/video.py

예시: lyrics.py (라인 20-30)

- async def get_store_default_info(conn: AsyncConnection):
+ def get_store_default_info(conn: Connection):
      query = """SELECT * FROM store_default_info;"""
-     result = await conn.execute(text(query))
+     result = conn.execute(text(query))
      return result.fetchall()

예시: INSERT 쿼리 (라인 360-400)

- async def insert_song_result(conn: AsyncConnection, params: dict):
+ def insert_song_result(conn: Connection, params: dict):
      insert_query = """INSERT INTO song_results_all (...) VALUES (...)"""
-     await conn.execute(text(insert_query), params)
-     await conn.commit()
+     conn.execute(text(insert_query), params)
+     conn.commit()

---

2.7 app/home/services/base.py - BaseService 클래스

파일: app/home/services/base.py

- from sqlalchemy.ext.asyncio import AsyncSession
+ from sqlalchemy.orm import Session

  class BaseService:
-     def __init__(self, model, session: AsyncSession):
+     def __init__(self, model, session: Session):
          self.model = model
          self.session = session

-     async def _get(self, id: UUID):
-         return await self.session.get(self.model, id)
+     def _get(self, id: UUID):
+         return self.session.get(self.model, id)

-     async def _add(self, entity):
+     def _add(self, entity):
          self.session.add(entity)
-         await self.session.commit()
-         await self.session.refresh(entity)
+         self.session.commit()
+         self.session.refresh(entity)
          return entity

-     async def _update(self, entity):
-         return await self._add(entity)
+     def _update(self, entity):
+         return self._add(entity)

-     async def _delete(self, entity):
-         await self.session.delete(entity)
+     def _delete(self, entity):
+         self.session.delete(entity)

---

3. 모델 파일 - 변경 불필요

다음 모델 파일들은 변경이 필요 없습니다:

• app/home/models.py
• app/lyric/models.py
• app/song/models.py
• app/video/models.py

모델 정의 자체는 비동기/동기와 무관하게 동일합니다. Mapped, mapped_column, relationship 등은 그대로 사용 가능합니다.

단, 관계 로딩 전략에서 lazy="selectin" 설정은 동기 환경에서도 작동하지만, 필요에 따라 lazy="joined" 또는 lazy="subquery"로 변경할 수 있습니다.

---

4. 수정 패턴 요약

4.1 Import 변경 패턴

# 엔진/세션 관련
- from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+ from sqlalchemy import create_engine
+ from sqlalchemy.orm import Session, sessionmaker

# 타입 힌트
- from typing import AsyncGenerator
+ from typing import Generator

# 컨텍스트 매니저
- from contextlib import asynccontextmanager
+ from contextlib import contextmanager

4.2 함수 정의 변경 패턴

- async def function_name(...):
+ def function_name(...):

4.3 await 제거 패턴

- result = await session.execute(query)
+ result = session.execute(query)

- await session.commit()
+ session.commit()

- await session.refresh(obj)
+ session.refresh(obj)

- await session.flush()
+ session.flush()

- await session.rollback()
+ session.rollback()

- await session.close()
+ session.close()

- await engine.dispose()
+ engine.dispose()

4.4 컨텍스트 매니저 변경 패턴

- async with SessionLocal() as session:
+ with SessionLocal() as session:

---

5. 영향받는 파일 목록

5.1 반드시 수정해야 하는 파일

파일	수정 범위
pyproject.toml	의존성 변경
config.py	DB URL 변경
app/database/session.py	전면 수정
app/database/session-prod.py	전면 수정
app/home/api/routers/v1/home.py	async/await 제거
app/lyric/api/routers/v1/lyric.py	async/await 제거
app/song/api/routers/v1/song.py	async/await 제거
app/video/api/routers/v1/video.py	async/await 제거
app/lyric/services/lyrics.py	async/await 제거
app/song/services/song.py	async/await 제거
app/video/services/video.py	async/await 제거
app/home/services/base.py	async/await 제거
app/song/dependencies.py	타입 힌트 변경
app/lyric/dependencies.py	타입 힌트 변경
app/video/dependencies.py	타입 힌트 변경
app/dependencies/database.py	타입 힌트 변경

5.2 수정 불필요한 파일

파일	이유
app/home/models.py	모델 정의는 동기/비동기 무관
app/lyric/models.py	모델 정의는 동기/비동기 무관
app/song/models.py	모델 정의는 동기/비동기 무관
app/video/models.py	모델 정의는 동기/비동기 무관

---

6. 주의사항

6.1 FastAPI와의 호환성

FastAPI는 동기 엔드포인트도 지원합니다. 동기 함수는 스레드 풀에서 실행됩니다:

# 동기 엔드포인트 - FastAPI가 자동으로 스레드풀에서 실행
@router.get("/items/{item_id}")
def get_item(item_id: int, session: Session = Depends(get_session)):
    return session.get(Item, item_id)

6.2 성능 고려사항

동기식으로 전환 시 고려할 점:

• 동시성 감소: 비동기 I/O의 이점 상실
• 스레드 풀 의존: 동시 요청이 많을 경우 스레드 풀 크기 조정 필요
• 블로킹 I/O: DB 쿼리 중 다른 요청 처리 불가

6.3 백그라운드 작업

현재 get_worker_session()으로 별도 이벤트 루프에서 실행되는 백그라운드 작업이 있습니다. 동기식 전환 시 스레드 기반 백그라운드 작업으로 변경해야 합니다:

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

def background_task():
    with get_worker_session() as session:
        # 작업 수행
        pass

# 실행
executor.submit(background_task)

---

7. 마이그레이션 단계

Step 1: 의존성 변경

1. pyproject.toml 수정
2. pip install pymysql 또는 uv sync 실행

Step 2: 설정 파일 수정

1. config.py의 DB URL 변경
2. app/database/session.py 전면 수정

Step 3: 라우터 수정

1. 각 라우터 파일의 async def → def 변경
2. 모든 await 키워드 제거
3. AsyncSession → Session 타입 힌트 변경

Step 4: 서비스 수정

1. 서비스 파일들의 async/await 제거
2. Raw SQL 쿼리 함수들 수정

Step 5: 의존성 수정

1. dependencies.py 파일들의 타입 힌트 변경

Step 6: 테스트

1. 모든 엔드포인트 기능 테스트
2. 성능 테스트 (동시 요청 처리 확인)

---

8. 결론

비동기에서 동기로 전환은 기술적으로 가능하지만, 다음을 고려해야 합니다:

장점:

• 코드 복잡도 감소 (async/await 제거)
• 디버깅 용이
• 레거시 라이브러리와의 호환성 향상

단점:

• 동시성 처리 능력 감소
• I/O 바운드 작업에서 성능 저하 가능
• FastAPI의 비동기 장점 미활용

현재 프로젝트가 FastAPI 기반이고 I/O 작업(DB, 외부 API 호출)이 많다면, 비동기 유지를 권장합니다. 동기 전환은 특별한 요구사항(레거시 통합, 팀 역량 등)이 있을 때만 고려하시기 바랍니다.