이코에코(Eco²) ORM Mapping Registry 통일 분석 리포트

작성일: 2026-01-07
목적: Clean Architecture 기반 장기 유지보수를 위한 ORM 매핑 전략 통일
Model: Opus 4.5
Agents: Cursor
PR: https://github.com/eco2-team/backend/pull/306

feat(apps): ORM Registry 통일 + 멀티스테이지 빌드 by mangowhoiscloud · Pull Request #306 · eco2-team/backend

 

---

1. 현재 상태 분석

1.1 도메인별 ORM 매핑 방식 비교

도메인	매핑 방식	Base/Registry 위치	도메인-ORM 결합도	start_mappers()
auth	Imperative	persistence_postgres/registry.py	🟢 분리됨	✅ start_all_mappers()
users	Imperative	각 mapping 파일 내	🟢 분리됨	✅ start_mappers()
character	Declarative	persistence_postgres/base.py	🔴 결합됨	❌ 없음 (암묵적)
location	Declarative	models.py 내	🔴 결합됨	❌ 없음 (암묵적)
character_worker	Raw SQL	N/A	🟢 분리됨	❌ N/A
users_worker	Raw SQL	N/A	🟢 분리됨	❌ N/A

1.2 현재 구조 상세

✅ auth (Imperative - 모범 사례)

apps/auth/infrastructure/persistence_postgres/
├── registry.py              # mapper_registry = registry()
├── mappings/
│   ├── __init__.py          # start_all_mappers()
│   ├── users.py             # Table(...) + map_imperatively()
│   ├── users_social_account.py
│   └── login_audit.py
├── adapters/
│   └── *.py
└── session.py

특징:

• 도메인 엔티티가 순수 Python 클래스 (Entity[UserId] 상속)
• ORM 결합은 map_imperatively()에서만 발생
• start_all_mappers() 호출로 초기화 시점 통제

✅ users (Imperative)

apps/users/infrastructure/persistence_postgres/
├── mappings/
│   ├── __init__.py          # start_mappers()
│   ├── user.py              # metadata + registry + Table + map_imperatively
│   ├── user_character.py
│   └── user_social_account.py
├── adapters/
│   └── *.py
├── constants.py
└── session.py

특징:

• 도메인 엔티티가 @dataclass (순수 Python)
• 각 파일마다 별도 metadata + registry 생성 (🔸 개선 필요)
• start_mappers() 호출로 초기화 시점 통제

⚠️ character (Declarative)

apps/character/infrastructure/persistence_postgres/
├── base.py                  # class Base(DeclarativeBase)
├── models/
│   ├── __init__.py
│   └── character.py         # class CharacterModel(Base) - ORM 모델
├── mappers.py               # model → entity 변환 함수
└── *.py                     # adapters

특징:

• CharacterModel(Base): ORM에 결합된 모델 클래스
• 도메인 엔티티 Character는 별도 @dataclass
• mappers.py에서 model → entity 수동 변환
• 문제: 모델 import 시점에 SQLAlchemy 메타데이터 암묵적 등록

⚠️ location (Declarative)

apps/location/infrastructure/persistence_postgres/
├── models.py                # Base + NormalizedLocationSite(Base)
└── location_reader_sqla.py  # 내부 변환

특징:

• models.py에 Base와 모델이 함께 정의
• 도메인 엔티티 NormalizedSite는 별도
• Reader 내부에서 _to_domain() 변환

---

2. 문제점 분석

2.1 Declarative 방식의 문제

Import Side-Effect

# character/infrastructure/persistence_postgres/models/character.py
from character.infrastructure.persistence_postgres.base import Base

class CharacterModel(Base):  # 👈 import 시점에 메타데이터 등록!
    __tablename__ = "characters"

영향:

• 테스트에서 격리 어려움
• Worker/CLI에서 예상치 못한 DB 연결 시도
• 마이그레이션 스크립트에서 충돌

ORM-Domain 결합

# Declarative: 모델 자체가 ORM에 결합
class CharacterModel(Base):
    id: Mapped[UUID] = mapped_column(...)  # SQLAlchemy 타입

# vs Imperative: 도메인은 순수
@dataclass
class User:
    id: UUID
    nickname: str | None

2.2 현재 혼합 상태의 리스크

1. 일관성 부재: 신규 개발자 온보딩 시 혼란
2. 테스트 복잡도: 각 도메인마다 다른 fixture 전략 필요
3. Alembic 통합: metadata 분산으로 마이그레이션 설정 복잡
4. 확장성: LLM/Scan 등 새 도메인 추가 시 어떤 패턴을 따를지 불명확

---

3. 통일 전략 제안

3.1 왜 Imperative로 통일해야 하는가

기준	Declarative	Imperative
도메인 순수성	❌ ORM 결합	✅ 완전 분리
Import side-effect	❌ 암묵적 등록	✅ 명시적 start_mappers()
테스트 격리	❌ 어려움	✅ clear_mappers() 가능
Value Object 지원	❌ 제한적	✅ 자유로운 매핑
도메인 이벤트	❌ 어려움	✅ 쉽게 통합
Clean Architecture	❌ 위반	✅ 준수

3.2 목표 구조

apps/<domain>/infrastructure/persistence_postgres/
├── registry.py              # 공유 또는 도메인별 registry
├── tables.py                # Table(...) 정의만
├── mappings.py              # map_imperatively() + start_mappers()
├── adapters/
│   ├── repository.py
│   └── unit_of_work.py
└── session.py

3.3 공용 Registry 옵션

# shared/infrastructure/persistence_postgres/registry.py
from sqlalchemy.orm import registry

mapper_registry = registry()
metadata = mapper_registry.metadata

장점: Alembic에서 단일 metadata 참조
단점: 서비스 간 의존성 발생

권장: 도메인별 registry 유지하되, Alembic 전용 통합 스크립트 작성

---

4. 도메인별 마이그레이션 계획

4.1 character (Declarative → Imperative)

Before

# models/character.py
class CharacterModel(Base):
    __tablename__ = "characters"
    id: Mapped[UUID] = mapped_column(...)

After

# tables.py
from sqlalchemy import Table, Column, String, Text
from character.infrastructure.persistence_postgres.registry import mapper_registry

characters_table = Table(
    "characters",
    mapper_registry.metadata,
    Column("id", UUID(as_uuid=True), primary_key=True),
    Column("code", String(64), unique=True, nullable=False),
    Column("name", Text, nullable=False),
    ...
    schema="character",
)

# mappings.py
from character.domain.entities import Character
from character.infrastructure.persistence_postgres.registry import mapper_registry
from character.infrastructure.persistence_postgres.tables import characters_table

def start_character_mapper() -> None:
    if hasattr(Character, "__mapper__"):
        return
    mapper_registry.map_imperatively(
        Character,
        characters_table,
    )

4.2 location (Declarative → Imperative)

Before

# models.py
class Base(DeclarativeBase): pass
class NormalizedLocationSite(Base): ...

After

# tables.py
location_sites_table = Table(
    "location_normalized_sites",
    mapper_registry.metadata,
    Column("positn_sn", BigInteger, primary_key=True),
    ...
    schema="location",
)

# mappings.py
from location.domain.entities import NormalizedSite
def start_location_mapper() -> None:
    mapper_registry.map_imperatively(NormalizedSite, location_sites_table)

4.3 auth (이미 Imperative - 정리만)

• 현재 구조 유지
• naming convention 통일: start_all_mappers() → start_mappers()

4.4 users (이미 Imperative - Registry 통합)

• 각 파일의 개별 metadata + registry → 공용 registry.py로 통합

---

5. 테스트 전략

5.1 Unit Test Fixture

# conftest.py
import pytest
from sqlalchemy.orm import clear_mappers

@pytest.fixture(autouse=True)
def clear_orm_mappers():
    yield
    clear_mappers()

@pytest.fixture
def setup_mappers():
    from character.infrastructure.persistence_postgres.mappings import start_mappers
    start_mappers()

5.2 Integration Test

@pytest.fixture(scope="session")
def db_session():
    from character.infrastructure.persistence_postgres.mappings import start_mappers
    start_mappers()
    # ... session setup

---

6. 결론

통일 목표

• Imperative Mapping으로 전체 통일
• 도메인 순수성 유지 (Clean Architecture 준수)
• 명시적 초기화 (start_mappers()) 패턴 표준화

예상 효과

1. 온보딩 단축: 일관된 ORM 패턴으로 학습 곡선 감소
2. 테스트 안정성: 격리된 테스트 환경
3. 확장성: 신규 도메인 (chat, scan-v2) 추가 시 명확한 템플릿
4. 유지보수성: import side-effect 제거로 예측 가능한 동작

예상 비용

• 개발: ~5일
• 리스크: 낮음 (기존 동작 유지, 구조만 변경)
• 테스트: 기존 테스트 수정 필요

---

부록: 표준 템플릿

registry.py

"""SQLAlchemy Mapper Registry."""
from sqlalchemy.orm import registry

mapper_registry = registry()

tables.py

"""Table Definitions."""
from sqlalchemy import Column, Table, String, Text
from sqlalchemy.dialects.postgresql import UUID
from <domain>.infrastructure.persistence_postgres.registry import mapper_registry

<entity>_table = Table(
    "<table_name>",
    mapper_registry.metadata,
    Column("id", UUID(as_uuid=True), primary_key=True),
    # ... columns
    schema="<schema>",
)

mappings.py

"""ORM Mappings."""
from <domain>.domain.entities import <Entity>
from <domain>.infrastructure.persistence_postgres.registry import mapper_registry
from <domain>.infrastructure.persistence_postgres.tables import <entity>_table


def start_<entity>_mapper() -> None:
    """<Entity> 매퍼 시작."""
    if hasattr(<Entity>, "__mapper__"):
        return
    mapper_registry.map_imperatively(<Entity>, <entity>_table)


def start_mappers() -> None:
    """모든 매퍼 시작."""
    start_<entity>_mapper()
    # ... other mappers