출처: Dev.to

다중 테넌트 계층에서 AI 기능 사용량을 집계할 때 SELECT N+1을 피하는 SQLAlchemy 하이브리드 프로퍼티

CitizenApp에서 3주 동안 성능 최적화에 매진했지만, 문제는 FastAPI 엔드포인트나 React 렌더링이 아니라 한 개의 대시보드 쿼리가 47,000개의 데이터베이스 호출을 발생시킨다는 것이었습니다. 원인? 테넌트 메트릭을 파이썬에서 계산했기 때문에 SQLAlchemy가 작업을 PostgreSQL로 넘기지 못한 것이었습니다.

대부분의 팀이 이 함정에 빠지는 이유는 직관적으로 더 쉽다고 느껴지기 때문입니다. 테넌트를 로드하고, 각 테넌트의 기능 사용량을 순회하며 메모리에서 채택률을 계산합니다. 코드가 자연스럽게 보이고, 개발 환경에서는 잘 동작합니다. 그런데 프로덕션에서 동시 사용자 500명을 맞이하면 데이터베이스 커넥션 풀이 금방 고갈됩니다.

이번 글에서는 SQLAlchemy의 hybrid_property 데코레이터가 왜 존재하는지, 다중 테넌트 시스템에 왜 필수적인지, 그리고 계산을 애플리케이션 레이어가 아니라 확장성이 뛰어난 데이터베이스 레이어로 옮기는 방법을 보여드리겠습니다.

---

문제: 애플리케이션 코드에서 계산된 메트릭

AI 기능 사용량을 다중 테넌트 계층 전체에 걸쳐 추적한다고 가정해봅시다. 데이터 모델은 대략 다음과 같습니다:

from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, func
from sqlalchemy.orm import relationship
from datetime import datetime

class Tenant(Base):
    __tablename__ = "tenants"
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    parent_id = Column(Integer, ForeignKey("tenants.id"), nullable=True)
    children = relationship("Tenant", remote_side=[id], backref="parent")
    ai_features = relationship("AIFeatureUsage", back_populates="tenant")

class AIFeatureUsage(Base):
    __tablename__ = "ai_feature_usage"
    id = Column(Integer, primary_key=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False)
    feature_name = Column(String, nullable=False)
    inference_cost = Column(Float, default=0.0)
    created_at = Column(DateTime, default=datetime.utcnow)
    tenant = relationship("Tenant", back_populates="ai_features")

Enter fullscreen mode
Exit fullscreen mode

대시보드 엔드포인트가 모든 테넌트를 로드하고 채택률(9개의 AI 기능 중 몇 개를 사용했는지)를 계산한다고 하면 코드는 다음과 같습니다:

# FastAPI endpoint — WRONG APPROACH
@app.get("/tenants/metrics")
async def get_tenant_metrics(session: Session = Depends(get_session)):
    tenants = session.query(Tenant).all()  # 1 query

    metrics = []
    for tenant in tenants:
        # 1 query per tenant to count features
        feature_count = session.query(AIFeatureUsage)\
            .filter(AIFeatureUsage.tenant_id == tenant.id)\
            .distinct(AIFeatureUsage.feature_name)\
            .count()

        adoption_rate = feature_count / 9
        metrics.append({
            "tenant_id": tenant.id,
            "adoption_rate": adoption_rate
        })

    return metrics

Enter fullscreen mode
Exit fullscreen mode

테넌트가 100개라면 총 101개의 쿼리가 실행됩니다. 1,000개라면 1,001개의 쿼리! 대시보드는 금방 사용 불가능해집니다.

---

해결책: 하이브리드 프로퍼티

SQLAlchemy의 hybrid_property를 사용하면 두 가지 상황에서 동작하는 계산된 속성을 정의할 수 있습니다:

• 인메모리 파이썬 표현식 – 인스턴스를 다룰 때
• SQL 표현식 – 쿼리 안에서 사용될 때

올바른 접근 방식은 다음과 같습니다:

from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import and_

class Tenant(Base):
    __tablename__ = "tenants"
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    parent_id = Column(Integer, ForeignKey("tenants.id"), nullable=True)
    children = relationship("Tenant", remote_side=[id], backref="parent")
    ai_features = relationship("AIFeatureUsage", back_populates="tenant")

    @hybrid_property
    def feature_adoption_rate(self) -> float:
        """
        Python-side: Count distinct features used by this tenant.
        """
        if not self.ai_features:
            return 0.0

        distinct_features = len(set(f.feature_name for f in self.ai_features))
        return distinct_features / 9

    @feature_adoption_rate.expression
    @classmethod
    def feature_adoption_rate(cls):
        """
        SQL-side: Compute adoption rate as a subquery.
        This runs in the database, not in Python.
        """
        from sqlalchemy.sql import select, func as sql_func

        feature_count = select(sql_func.count(
            sql_func.distinct(AIFeatureUsage.feature_name)
        )).where(
            AIFeatureUsage.tenant_id == cls.id
        ).correlate(cls).scalar_subquery()

        return feature_count / 9

Enter fullscreen mode
Exit fullscreen mode

이제 엔드포인트는 이렇게 간단해집니다:

@app.get("/tenants/metrics")
async def get_tenant_metrics(session: Session = Depends(get_session)):
    tenants = session.query(Tenant).add_columns(
        Tenant.feature_adoption_rate.label("adoption_rate")
    ).all()

    return [
        {
            "tenant_id": t.id,
            "adoption_rate": t.adoption_rate
        }
        for t in tenants
    ]

Enter fullscreen mode
Exit fullscreen mode

단 한 번의 쿼리만 실행됩니다. 단일 JOIN으로 데이터베이스 레이어에서 채택률을 계산하므로 성능이 크게 개선됩니다.

---

다중 테넌트 계층: 트리 상위로 계산된 값 전파하기

실제 다중 테넌트 시스템은 계층 구조를 가집니다. 상위 테넌트는 하위 테넌트들의 집계 사용량을 물려받습니다. 이때 하이브리드 프로퍼티가 특히 유용합니다:

class Tenant(Base):
    __tablename__ = "tenants"
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    parent_id = Column(Integer, ForeignKey("tenants.id"), nullable=True)
    children = relationship("Tenant", remote_side=[id], backref="parent")
    ai_features = relationship("AIFeatureUsage", back_populates="tenant")

    @hybrid_property
    def total_inference_cost(self) -> float:
        """
        Include costs from this tenant + all descendants.
        """
        own_cost = sum(f.inference_cost for f in self.ai_features)
        children_cost = sum(c.total_inference_cost for c in self.children)
        return own_cost + children_cost

    @total_inference_cost.expression
    @classmethod
    def total_inference_cost(cls):
        """
        Recursive CTE in SQL (PostgreSQL 12+).
        """
        from sqlalchemy import text

        # Direct costs for this tenant
        direct = select(func.coalesce(
            func.sum(AIFeatureUsage.inference_cost), 0
        )).where(
            AIFeatureUsage.tenant_id == cls.id
        ).correlate(cls).scalar_subquery()

        # Children costs (you'd typically use a recursive CTE for large trees)
        # Simplified here for clarity
        children_costs = select(func.coalesce(
            func.sum(Tenant.total_inference_cost), 0
        )).where(
            Tenant.parent_id == cls.id
        ).correlate(cls).scalar_subquery()

        return direct + children_costs

Enter fullscreen mode
Exit fullscreen mode

---

함정: 하이브리드 프로퍼티가 조용히 실패할 때

제가 겪은 가장 큰 실수는 하이브리드 프로퍼티가 SQL로 변환되지 않을 때였습니다. 일부 표현식은 너무 복잡하거나 파이썬 전용 로직을 사용하기 때문에 SQL로 변환되지 않습니다:

class Tenant(Base):
    @hybrid_property
    def permission_inheritance_depth(self) -> int:
        """
        Count how many levels deep in the hierarchy.
        This LOOKS like it should work...
        """
        if self.parent:
            ret

(코드가 중간에 끊겼지만, 핵심은 파이썬 전용 로직(if self.parent: ret)이 포함돼 있으면 SQL 변환이 불가능하다는 점입니다.)

교훈

1. 가능하면 순수 SQL 함수와 연산만 사용하세요.
2. 복잡한 파이썬 로직이 필요하면 별도 서브쿼리/CTE로 분리하거나, 데이터베이스 함수(예: PostgreSQL PL/pgSQL)로 옮기세요.
3. hybrid_property를 정의한 뒤 **<property>.expression**이 실제로 SQLAlchemy가 생성한 SQL을 확인해 보세요. str(query) 혹은 `query.statement.compile(dialect=