ChangeRequest
  ├─ target_type (SERVICE, WORKFLOW, DASHBOARD, ...)
  ├─ target_version_id ──→ Version (ServiceVersion, WorkflowVersion, ...)
  ├─ submitter_id ──→ User
  └─ reviewer_id  ──→ User

Version ──→ Parent (Service, Workflow, Dashboard, ...)
Parent  ──→ TargetMeta ──→ ProjectSpace

@event.listens_for(engine.sync_engine, "before_cursor_execute")
def _before_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault("_query_start", []).append(time.perf_counter())

@event.listens_for(engine.sync_engine, "after_cursor_execute")
def _after_execute(conn, cursor, statement, parameters, context, executemany):
    elapsed_ms = round(
        (time.perf_counter() - conn.info["_query_start"].pop()) * 1000, 3
    )
    _sql_profiles.append({
        "ms": elapsed_ms,
        "sql": _inline_params(statement, parameters),
    })

def _inline_params(stmt: str, params) -> str:
    """바인드 파라미터를 SQL에 직접 삽입하여 실행 가능한 쿼리 생성."""
    if isinstance(params, dict):
        escaped = {k: _escape(v) for k, v in params.items()}
        return re.sub(r"%\((\w+)\)s", lambda m: escaped.get(m.group(1), "NULL"), stmt)
    # ... tuple/list 처리

@contextmanager
def mysql_sql_no_cache(engine):
    def _inject(conn, cursor, statement, parameters, context, executemany):
        stripped = statement.lstrip()
        if not stripped.upper().startswith("SELECT "):
            return statement, parameters
        return f"SELECT SQL_NO_CACHE {stripped[7:]}", parameters

    event.listen(sync_engine, "before_cursor_execute", _inject, retval=True)
    try:
        yield
    finally:
        event.remove(sync_engine, "before_cursor_execute", _inject)

======================================================================
  SQL PROFILE REPORT — 144 queries, 5807.37ms total
======================================================================

  [10] 🐢 SLOW  1677.123ms
  ------------------------------------------------------------------
  SELECT SQL_NO_CACHE count(*) FROM (SELECT ... ) AS anon_1

  [11] 🐢 SLOW  1666.456ms
  ------------------------------------------------------------------
  SELECT SQL_NO_CACHE anon_1.approval_id FROM (SELECT ... ) AS anon_1 ...

base_query = (
    select(
        ChangeRequest.id.label('request_id'),
        # ...
        ProjectSpace.name.label('project_space_name'),  # ← 이것 때문에 join 필요
    )
    # ... 12개 revision/parent outerjoin ...
    .outerjoin(
        TargetMeta,
        or_(
            and_(TargetMeta.target_type == 'SERVICE',
                 TargetMeta.target_version_id == ServiceVersion.service_id),
            and_(TargetMeta.target_type == 'WORKFLOW',
                 TargetMeta.target_version_id == WorkflowVersion.workflow_id),
            # ... 4개 더
        )
    )
    .outerjoin(ProjectSpace, ProjectSpace.id == TargetMeta.project_space_id)
)

query = select(
    ChangeRequest,
    # ...
    ServiceVersion,          # ← ORM 객체
    AutomationRelease,       # ← ORM 객체
    Automation,              # ← ORM 객체
)

for request, ..., project_space_name in rows:
    if not is_admin:
        # 캐시 미스마다 DB 조회 발생
        updatable = await permission_service.can_user_create_resource(
            resource_type=request.target_type,
            project_space_name=project_space_name,
            ...
        )

class QueryCounter:
    """SQLAlchemy 엔진의 쿼리 수와 총 소요시간을 측정하는 컨텍스트 매니저."""

    def __init__(self, engine):
        self.engine = engine.sync_engine
        self.queries = []

    def _before_execute(self, conn, cursor, stmt, params, context, executemany):
        conn.info.setdefault("_qc_start", []).append(time.perf_counter())

    def _after_execute(self, conn, cursor, stmt, params, context, executemany):
        start = conn.info["_qc_start"].pop()
        self.queries.append({'time': (time.perf_counter() - start) * 1000})

    def __enter__(self):
        self.queries.clear()
        event.listen(self.engine, "before_cursor_execute", self._before_execute)
        event.listen(self.engine, "after_cursor_execute", self._after_execute)
        return self

    def __exit__(self, *args):
        event.remove(self.engine, "before_cursor_execute", self._before_execute)
        event.remove(self.engine, "after_cursor_execute", self._after_execute)

    @property
    def count(self): return len(self.queries)

    @property
    def total_time_ms(self): return sum(q['time'] for q in self.queries)

async def test_admin_default_query_count(self, ...):
    with QueryCounter(engine) as qc:
        total, approvals = await list_change_requests(user=admin, pg=1, pg_size=10)

    assert total == 1000
    assert qc.count <= 25      # 쿼리 수 상한
    assert qc.total_time_ms < 500  # 시간 상한 (ms)

async def test_non_admin_default_query_count(self, ...):
    with QueryCounter(engine) as qc:
        total, _ = await list_change_requests(user=non_admin, pg=1, pg_size=10)

    assert total == 500
    assert qc.count <= 35
    assert qc.total_time_ms < 500

async def test_order_by_id_desc(self, ...):
    """ID 내림차순 정렬이 올바르게 동작하는지 검증."""
    await create_multiple_change_requests(
        session, admin, project_space, base_resources,
        count=10, prefix='order-test'
    )

    total, responses = await list_change_requests(
        user=admin, pg=1, pg_size=10,
        order_field=OrderField.id, order_type=OrderType.desc,
        session=session,
    )

    ids = [r.id for r in responses]
    assert ids == sorted(ids, reverse=True), "IDs should be in descending order"

async def test_order_by_target_type(self, ...):
    """target_type 정렬: asc와 desc가 서로 역순인지 검증."""
    await create_multiple_change_requests(
        session, admin, project_space, base_resources,
        count=12, prefix='order-test'
    )

    _, responses_asc = await list_change_requests(
        user=admin, pg=1, pg_size=20,
        order_field=OrderField.target_type, order_type=OrderType.asc,
        session=session,
    )
    _, responses_desc = await list_change_requests(
        user=admin, pg=1, pg_size=20,
        order_field=OrderField.target_type, order_type=OrderType.desc,
        session=session,
    )

    types_asc = [r.target_type for r in responses_asc]
    types_desc = [r.target_type for r in responses_desc]
    assert types_asc == list(reversed(types_desc))

base_query = (
    select(
        ChangeRequest.id.label('request_id'),
        # ...
        ProjectSpace.name.label('project_space_name'),
    )
    # ... revision/parent outerjoins ...
    .outerjoin(
        TargetMeta,
        or_(
            and_(TargetMeta.target_type == 'SERVICE',
                 TargetMeta.target_version_id == ServiceVersion.service_id),
            # ... 5개 더
        )
    )
    .outerjoin(ProjectSpace, ProjectSpace.id == TargetMeta.project_space_id)
    .where(...)
)

base_query = (
    select(
        ChangeRequest.id.label('request_id'),
        # ...
        # project_space_name 제거됨 — Phase 3에서만 필요
    )
    # ... revision/parent outerjoins 유지 ...
    # target_meta_tb, project_space_tb는 Phase 3에서만 사용
    .where(...)
)

query = select(
    ChangeRequest,
    # ...
    ServiceVersion,          # ORM 객체 전체 → selectin cascade 유발
    AutomationRelease,       # ORM 객체 전체
    Automation,              # ORM 객체 전체
)

for request, ..., svc_version, auto_release, automation in rows:
    if request.target_type == 'AUTOMATION':
        parent_id = auto_release.automation_id
        parent_name = automation.name
    # ...
    if svc_version is None:
        raise ...
    data.update(is_external=svc_version.is_external)

query = select(
    ChangeRequest,
    # ...
    # 스칼라 컬럼만 조회 — ORM 로딩 방지
    ServiceVersion.is_external.label('is_external'),
    AutomationRelease.automation_id.label('auto_parent_id'),
    Automation.name.label('auto_name'),
)

for request, ..., is_external, auto_parent_id, auto_name in rows:
    if request.target_type == 'AUTOMATION':
        parent_id = auto_parent_id
        parent_name = auto_name
    # ...
    if is_external is None:
        raise ...
    data.update(is_external=is_external)

# 각 고유 (target_type, project_space_name) 쌍마다 DB 조회
updatable = await permission_service.can_user_create_resource(
    resource_type=request.target_type,
    user=user,
    project_space_name=project_space_name,
    session=session     # ← 내부에서 get_user_statements() DB 호출
)

# 함수 시작 시 이미 조회한 user_statements 재사용
updatable = permission_service.can_user_create_resource_from_statements(
    statements=user_statements,   # ← 사전 조회된 데이터
    resource_type=request.target_type,
    project_space_name=project_space_name,
)

사전 설정 (46%) ████████████████████████
Phase 1+2  (9%) █████
Phase 3   (13%) ███████
User 관계 로딩 (31%) ████████████████