다차원 검색을 수행하는 데이터베이스 (3) FastAPI 게이트웨이

설정 파일

Python 게이트웨이를 작성할 디렉토리 및 파일을 생성한다.

~/workspace/emotion-dict$ mkdir python-gateway && cd python-gateway
~/workspace/emotion-dict/python-gateway$ uv init
~/workspace/emotion-dict/python-gateway$ uv add fastapi uvicorn grpcio grpcio-tools python-dotenv

의존성 파일은 자동으로 생성되며 설명 정도만 추가로 수정해 주면 된다.

python-gateway/pyproject.toml

[project]
name = "python-gateway"
version = "0.1.0"
description = "다차원 감정 사전 검색을 위한 FastAPI 기반 API 게이트웨이"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.136.3",
    "grpcio>=1.80.0",
    "grpcio-tools>=1.80.0",
    "python-dotenv>=1.2.2",
    "uvicorn>=0.47.0",
]

구조화된 프로젝트를 위해 하위 디렉토리를 생성하고
Python 패키지로 인식되도록 __init__.py 파일을 생성한다.

~/workspace/emotion-dict/python-gateway$ mkdir -p app/core app/schemas app/api app/grpc_client
~/workspace/emotion-dict/python-gateway$ touch app/__init__.py app/core/__init__.py app/schemas/__init__.py app/api/__init__.py app/grpc_client/__init__.py

🤖 AI AGENT | __init__.py

각 디렉토리에 __init__.py 을 추가하는 것은 문법적으로는 더 이상 필수가 아니지만, 실무(Production) 환경의 백엔드 아키텍처에서는 여전히 생성하는 것이 정석(Best Practice)입니다.

그 이유를 파이썬의 동작 방식(PEP 420)과 실무적인 관점으로 나누어 명확히 설명해 드리겠습니다.

🔍 1. 문법적 관점: 네임스페이스 패키지 (PEP 420)

과거 파이썬 3.2 이하 버전에서는 디렉토리에 __init__.py가 없으면 파이썬이 이를 패키지로 인식하지 못해 ImportError를 발생시켰습니다. (이를 Regular Package라고 부릅니다.)

하지만 파이썬 3.3부터 PEP 420 (Implicit Namespace Packages)이 도입되면서, __init__.py가 없는 디렉토리도 자동으로 패키지로 인식하고 임포트할 수 있게 되었습니다. 즉, 현대 파이썬에서는 단순히 모듈을 불러오기 위한 목적이라면 __init__.py를 생략해도 코드는 정상적으로 동작합니다.

🏛️ 2. 실무적 관점: 그럼에도 불구하고 왜 생성해야 하는가? (Why?)

현대적인 프레임워크(FastAPI 등)를 사용하는 실무 환경에서 우리가 모든 디렉토리에 __init__.py를 명시적으로 추가하는 이유는 다음과 같습니다.

1. 명시성 (Explicit is better than implicit):
   디렉토리에 __init__.py가 존재한다는 것 자체로, 해당 폴더가 단순한 정적 자원(데이터 파일, 템플릿 등)을 모아둔 폴더가 아니라 "파이썬 코드 패키지"임을 개발자와 시스템에게 명확하게 선언하는 문서 역할을 합니다.
2. 정적 분석 도구 및 타입 체커 호환성:
   실무에서 필수적으로 사용하는 mypy, pyright, pytest, flake8 등의 도구들은 __init__.py가 없는 네임스페이스 패키지 환경에서 간혹 경로를 제대로 해석하지 못하거나 엣지 케이스(Edge Case) 버그를 일으키는 경우가 있습니다. 빈 __init__.py 하나가 CI/CD 파이프라인에서의 알 수 없는 모듈 인식 실패를 방지하는 훌륭한 보험이 됩니다.
3. 패키지 초기화 및 인터페이스 캡슐화:
   바로 앞 단계에서 우리가 app/grpc_client/__init__.py에 sys.path 패치와 __all__을 작성했던 것을 기억하시나요? 모듈 내부의 복잡한 구조를 숨기고, 외부에는 필요한 클래스나 함수만 깔끔하게 노출시키기 위한 진입점(Entry Point)으로서 __init__.py는 여전히 필수적인 역할을 수행합니다.

💡 백엔드 마스터의 결론:
"작은 스크립트를 작성할 때는 생략해도 무방하지만, 구조화된 마이크로서비스나 라이브러리를 구축할 때는 빈 파일(touch __init__.py)이라도 명시적으로 생성하여 모듈의 경계를 확정 짓는 것이 가장 안전하고 전문적인 방식입니다."

proto/emotion_search.proto 파일에 정의한 gRPC 서비스 명세를 기반으로
Python 파일을 생성하는 명령어를 실행한다.

~/workspace/emotion-dict/python-gateway$ uv run python -m grpc_tools.protoc \
	-I../proto \
    --python_out=./app/grpc_client \
    --grpc_python_out=./app/grpc_client \
    ../proto/emotion_search.proto

emotion_search_pb2.py 파일과 emotion_search_pb2_grpc.py 파일이
python-gateway/app/grpc_client 디렉토리 내에 자동 생성된다.

생성된 파일들이 서로를 문제없이 찾을 수 있도록,
grpc_client/__init__.py 파일에 다음 코드를 작성한다.

python-gateway/app/grpc_client/__init__.py

import sys
from pathlib import Path

current_dir = Path(__file__).resolve().parent
if str(current_dir) not in sys.path:
    sys.path.append(str(current_dir))

from . import emotion_search_pb2
from . import emotion_search_pb2_grpc

__all__ = ["emotion_search_pb2", "emotion_search_pb2_grpc"]

코드 작성

코어 설정

환경변수를 로드하고 로깅을 설정하는 코드를 작성한다.

python-gateway/app/core/config.py

import os
import logging
from dotenv import load_dotenv
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent.parent.parent
load_dotenv(dotenv_path=BASE_DIR / ".env")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("gateway")

GRPC_SERVER_ADDR = os.getenv("GRPC_HOST", "127.0.0.1:50051")

서버 생명주기와 커스텀 응답 클래스를 관리하는 코드를 작성한다.

python-gateway/app/core/server.py

import grpc
from contextlib import asynccontextmanager
from fastapi import FastAPI

from app.core.config import logger, GRPC_SERVER_ADDR
from app.grpc_client import emotion_search_pb2_grpc

# gRPC 연결 상태 관리를 위한 전역 딕셔너리
grpc_context = {"channel": None, "stub": None}

@asynccontextmanager
async def lifespan(app: FastAPI):
    """서버 시작 시 gRPC 채널을 열고, 종료 시 안전하게 닫습니다."""
    logger.info(f"🚀 게이트웨이 시작: Rust 엔진({GRPC_SERVER_ADDR}) 연결 중...")
   
    channel = grpc.aio.insecure_channel(GRPC_SERVER_ADDR)
    grpc_context["channel"] = channel
    grpc_context["stub"] = emotion_search_pb2_grpc.EmotionSearchServiceStub(channel)
   
    yield

    logger.info("🛑 종료 신호 감지: gRPC 채널을 안전하게 닫습니다...")
    await channel.close()
    logger.info("✅ 리소스 정리 완료.")

데이터 스키마

검색 쿼리에 대한 데이터 스키마를 작성한다.

python-gateway/app/schemas/emotion.py

from pydantic import BaseModel
from typing import List

class VectorSearchReq(BaseModel):
    target_vector: List[float]
    limit: int = 1000

class TaxonomySearchReq(BaseModel):
    path_query: str
    limit: int = 1000

class TextSearchReq(BaseModel):
    query: str
    limit: int = 1000

라우팅

실제적인 엔트리포인트를 작성한다.

python-gateway/app/api/emotion.py

import grpc
from fastapi import APIRouter, HTTPException

from app.schemas.emotion import VectorSearchReq, TaxonomySearchReq, TextSearchReq
from app.grpc_client import emotion_search_pb2
from app.core.server import grpc_context
from app.core.config import logger

router = APIRouter(prefix="/api/v1/search", tags=["Emotion Search"])

@router.post("/vector", summary="2차원 백테(VA) 기반 유사도 검색")
async def search_by_vector(req: VectorSearchReq):
    try:
        grpc_req = emotion_search_pb2.VectorSearchRequest(
            target_vector=req.target_vector, limit=req.limit
        )
        res = await grpc_context["stub"].SearchByVector(grpc_req)

        return {
            "emotions": [
                {
                    "id": e.id,
                    "word": e.word,
                    "definition": e.definition,
                    "taxonomy_path": e.taxonomy_path,
                    "va_vector": list(e.va_vector)
                } for e in res.emotions
            ]
        }
    except grpc.aio.AioRpcError as e:
        logger.error(f"gRPC Vector 통신 장애: {e.details()}")
        raise HTTPException(status_code=500, detail="엔진 서버 오류")

@router.post("/taxonomy", summary="계층 분류(ltree) 검색")
async def search_by_taxonomy(req: TaxonomySearchReq):
    try:
        grpc_req = emotion_search_pb2.TaxonomySearchRequest(
            path_query=req.path_query, limit=req.limit
        )
        res = await grpc_context["stub"].SearchByTaxonomy(grpc_req)

        return {
            "emotions": [
                {
                    "id": e.id,
                    "word": e.word,
                    "definition": e.definition,
                    "taxonomy_path": e.taxonomy_path,
                    "va_vector": list(e.va_vector)
                } for e in res.emotions
            ]
        }
    except grpc.aio.AioRpcError as e:
        logger.error(f"gRPC Taxonomy 통신 장애: {e.details()}")
        raise HTTPException(status_code=500, detail="엔진 서버 오류")

@router.post("/text", summary="사전적 의미 풀텍스트 검색")
async def search_by_text(req: TextSearchReq):
    try:
        grpc_req = emotion_search_pb2.TextSearchRequest(
            query=req.query, limit=req.limit
        )
        res = await grpc_context["stub"].SearchByText(grpc_req)

        return {
            "emotions": [
                {
                    "id": e.id,
                    "word": e.word,
                    "definition": e.definition,
                    "taxonomy_path": e.taxonomy_path,
                    "va_vector": list(e.va_vector)
                } for e in res.emotions
            ]
        }
    except grpc.aio.AioRpcError as e:
        logger.error(f"gRPC Text 통신 장애: {e.details()}")
        raise HTTPException(status_code=500, detail="엔진 서버 오류")

main.py

API 라우터를 등록한다.

python-gateway/main.py

from fastapi import FastAPI
from app.core.server import lifespan, UTF8ORJSONResponse
from app.api.emotion import router as emotion_router

app = FastAPI(
    title="다차원 감정 검색 API Gateway",
    description="Python FastAPI ↔ Rust gRPC 엔진",
    lifespan=lifespan
)

app.include_router(emotion_router)

실행 및 테스트

지금까지 작성한 Python 게이트웨이의 구조는 다음과 같다.

~/workspace/emotion-dict/python-gateway$ tree     
.
├── README.md
├── app
│   ├── __init__.py
│   ├── api
│   │   ├── __init__.py
│   │   └── emotion.py
│   ├── core
│   │   ├── __init__.py
│   │   ├── config.py
│   │   └── server.py
│   ├── grpc_client
│   │   ├── __init__.py
│   │   ├── emotion_search_pb2.py
│   │   └── emotion_search_pb2_grpc.py
│   └── schemas
│       ├── __init__.py
│       └── emotion.py
├── main.py
├── pyproject.toml
└── uv.lock

6 directories, 15 files

Rust 엔진 실행 중인 상태로 다음을 실행한다.

~/workspace/emotion-dict/python-gateway$ uv run uvicorn main:app --host 0.0.0.0 --port 8000

AI가 작성해준 테스트 스크립트를 사용하겠다.

scripts/test/test_gateway.py

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "httpx",
#     "rich",
# ]
# ///

import asyncio
import httpx
from rich.console import Console
from rich.table import Table

console = Console()
BASE_URL = "http://127.0.0.1:8000/api/v1/search"

async def test_vector_search(client: httpx.AsyncClient):
    console.print("\n[bold cyan]1. 벡터(VA) 기반 K-NN 검색 테스트[/bold cyan]")
    payload = {"target_vector": [0.8, 0.8], "limit": 3}

    response = await client.post(f"{BASE_URL}/vector", json=payload)
    response.raise_for_status()
    data = response.json()

    table = Table(show_header=True, header_style="bold magenta")
    table.add_column("단어")
    table.add_column("V (정서가)")
    table.add_column("A (각성가)")
    table.add_column("분류 경로")

    for item in data.get("emotions", []):
        word = item["word"]
        v, a = item["va_vector"]
        path = item["taxonomy_path"]
        table.add_row(word, f"{v:.3f}", f"{a:.3f}", path)

    console.print(table)

async def test_taxonomy_search(client: httpx.AsyncClient):
    console.print("\n[bold cyan]2. 계층 분류(ltree) 기반 검색 테스트[/bold cyan]")
    payload = {"path_query": "negative.low_arousal.*", "limit": 3}

    response = await client.post(f"{BASE_URL}/taxonomy", json=payload)
    response.raise_for_status()
    data = response.json()

    for item in data.get("emotions", []):
        console.print(f" - [bold]{item['word']}[/bold] ({item['path']})")

async def test_text_search(client: httpx.AsyncClient):
    console.print("\n[bold cyan]3. 텍스트(FTS) 의미 기반 검색 테스트[/bold cyan]")
    payload = {"query": "마음이", "limit": 2}

    response = await client.post(f"{BASE_URL}/text", json=payload)
    response.raise_for_status()
    data = response.json()

    for item in data.get("emotions", []):
        console.print(f" - [bold]{item['word']}[/bold]: {item['definition']}")

async def main():
    console.print("[bold yellow]🚀 FastAPI 게이트웨이 통합 테스트 시작...[/bold yellow]")
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            await test_vector_search(client)
            await test_taxonomy_search(client)
            await test_text_search(client)
        console.print("\n[bold green]✅ 모든 API 테스트가 성공적으로 완료되었습니다![/bold green]")
    except httpx.ConnectError:
        console.print("[bold red]❌ 연결 실패: FastAPI 서버가 실행 중인지 확인하세요.[/bold red]")
    except Exception as e:
        console.print(f"[bold red]❌ 테스트 중 오류 발생: {e}[/bold red]")

if __name__ == "__main__":
    asyncio.run(main())

이 테스트 스크립트를 실행하면,

 ~/workspace/emotion-dict/python-gateway$ uv run scripts/test/test_gateway.py
🚀 FastAPI 게이트웨이 통합 테스트 시작...

1. 벡터(VA) 기반 K-NN 검색 테스트
┏━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━┓
┃ 단어       ┃ V (정서가) ┃ A (각성가) ┃ 분류 경로             ┃
┡━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━┩
│ 신바람나다 │ 0.180      │ 0.175      │ positive.high_arousal │
│ 신나다     │ 0.240      │ 0.222      │ positive.high_arousal │
│ 통쾌하다   │ 0.198      │ 0.177      │ positive.high_arousal │
└────────────┴────────────┴────────────┴───────────────────────┘

2. 계층 분류(ltree) 기반 검색 테스트
 - 가뜬하다 (negative.low_arousal)
 - 가련하다 (negative.low_arousal)
 - 가소롭다 (negative.low_arousal)

3. 텍스트(FTS) 의미 기반 검색 테스트
 - 가뜬하다: 마음이 가볍고 상쾌하다. ‘가든하다’보다 센 느낌을 준다.
 - 가엾다: 마음이 아플 만큼 안되고 처연하다.

✅ 모든 API 테스트가 성공적으로 완료되었습니다!