계층 구조를 재귀적으로 탐색하는 데이터베이스 (3) FastAPI 게이트웨이

설정 파일

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

~/workspace/occult-graph$ mkdir python-gateway && cd python-gateway
~/workspace/occult-graph/python-gateway$ uv init
~/workspace/occult-graph/python-gateway$ uv add fastapi uvicorn grpcio grpcio-tools python-dotenv pydantic-settings

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

python-gateway/pyproject.toml

[project]
name = "python-gateway"
version = "0.1.0"
description = "Rust gRPC 코어 엔진과 통신하여 데이터를 서빙하는 FastAPI 기반 오컬트 지식 그래프 게이트웨이"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.136.3",
    "grpcio>=1.81.0",
    "grpcio-tools>=1.81.0",
    "pydantic-settings>=2.14.1",
    "python-dotenv>=1.2.2",
    "uvicorn>=0.48.0",
]

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

~/workspace/emotion-dict/python-gateway$ mkdir -p app/pb
~/workspace/emotion-dict/python-gateway$ touch app/__init__.py
~/workspace/emotion-dict/python-gateway$ uv run python -m grpc_tools.protoc \
	-I../proto \
    --python_out=./app/pb \
    --grpc_python_out=./app/pb \
    ../proto/occult.proto

코드 작성

환경 변수 검증

환경변수를 로드하는 설정 모듈을 작성한다.

python-gateway/app/config.py

from pydantic_settings import BaseSettings, SettingsConfigDict
from pathlib import Path

class Settings(BaseSettings):
 	# 초기화하지 않은 값은 .env 파일에 해당 값이 없을 때 Pydantic이 즉시 ValidationError를 발생시키며 앱 종료
    GRPC_HOST: str

    FRONTEND_URL: str = ""

    model_config = SettingsConfigDict(
        env_file=str(Path(__file__).resolve().parents[2] / ".env"),
        env_file_encoding="utf-8",
        extra="ignore"
    )

settings = Settings()

gRPC 통신 계층

비동기 gRPC 채널을 관리하는 핵심 모듈을 작성한다.

python-gateway/app/grpc_client.py

import grpc
from fastapi import HTTPException, status
import logging
import sys
from pathlib import Path

pb_path = Path(__file__).parent / "pb"
sys.path.append(str(pb_path))

import occult_pb2
import occult_pb2_grpc
from app.config import settings

logger = logging.getLogger("fastapi_gateway")

class OccultGrpcClient:
    def __init__(self):
        self.target = f"{settings.GRPC_HOST}"
        self.channel = None
        self.stub = None

    def connect(self):
        self.channel = grpc.aio.insecure_channel(self.target)
        self.stub = occult_pb2_grpc.OccultKnowledgeStub(self.channel)
        logger.info(f"gRPC Client가 Rust 코어 엔진({self.target}과 연결되었습니다.")

    async def close(self):
        await self.channel.close()
        logger.info("gRPC 채널이 안전하게 닫혔습니다.")

grpc_client = OccultGrpcClient()

async def get_grpc_client() -> OccultGrpcClient:
    return grpc_client

def handle_grpc_error(e: grpc.aio.AioRpcError):
    if e.code() == grpc.StatusCode.NOT_FOUND:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=e.details())
    elif e.code() == grpc.StatusCode.INVALID_ARGUMENT:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=e.details())
    else:
        logger.error(f"Rust 엔진 통신 에러: {e.details()}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="내부 엔진 오류")

API 라우터

python-gateway/app/main.py

import json
import logging
from contextlib import asynccontextmanager
from typing import List, Dict, Any

from fastapi import FastAPI, Depends, Query
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.grpc_client import grpc_client, get_grpc_client, handle_grpc_error, OccultGrpcClient

import sys
from pathlib import Path

pb_path = Path(__file__).parent / "pb"
sys.path.append(str(pb_path))

import occult_pb2

logging.basicConfig(level=logging.INFO)

@asynccontextmanager
async def lifespan(app: FastAPI):
	grpc_client.connect()
    yield
    await grpc_client.close()

app = FastAPI(
    title="Occult Knowledge Graph Gateway",
    description="Rust gRPC 엔진을 매개하는 FastAPI 게이트웨이",
    version="1.0.0",
    lifespan=lifespan
)

origins = [
    "http://localhost:5173",
    "http://127.0.0.1:5173",
    "http://localhost:3000",
]

if settings.FRONTEND_URL and settings.FRONTEND_URL not in origins:
    origins.append(settings.FRONTEND_URL)

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

def serialize_node(node: occult_pb2.Node) -> Dict[str, Any]:
    """Protobuf Node 객체를 Python 딕셔너리로 변환하는 헬퍼 함수"""
    return {
        "id": node.id,
        "name": node.name,
        "entity_type": node.entity_type,
        "attributes": json.loads(node.attributes_json) if node.attributes_json else {}
    }

def serialize_path(path: occult_pb2.EdgePath) -> Dict[str, Any]:
    """Protobuf EdgePath 객체를 Python 딕셔너리로 변환하는 헬퍼 함수"""
    return {
        "parent_id": path.parent_id,
        "child_id": path.child_id,
        "relation_type": path.relation_type,
        "depth": path.depth,
        "weight": path.weight
    }

@app.get("/api/nodes/search", summary="노드 검색", response_model=List[Dict[str, Any]])
async def search_nodes(
    query: str = Query(..., description="검색어 (이름 및 속성 내부 검색)"),
    entity_type: str = Query("", description="특정 엔티티 타입 필터 (선택 사항)"),
    limit: int = Query(10, ge=1, le=100, description="최대 반환 개수"),
    client: OccultGrpcClient = Depends(get_grpc_client)
):
    try:
        request = occult_pb2.SearchNodesRequest(
            query=query,
            entity_type_filter=entity_type,
            limit=limit
        )

        response = await client.stub.SearchNodes(request)

        return [serialize_node(node) for node in response.nodes]
    except grpc.aio.AioRpcError as e:
        handle_grpc_error(e)

@app.get("/api/nodes/{identifier}", summary="단일 노드 상세 조회", response_model=Dict[str, Any])
async def get_node(
    identifier: str,
    client: OccultGrpcClient = Depends(get_grpc_client)
):
    try:
        request = occult_pb2.GetNodeRequest(
            identifier=identifier
        )
      
        response = await client.stub.GetNode(request)

        return serialize_node(response.node)
    except grpc.aio.AioRpcError as e:
        handle_grpc_error(e)

@app.get("/api/graph/traverse", summary="지식 그래프 재귀 탐색", response_model=Dict[str, Any])
async traverse_graph(
    start_node: str = Query(..., description="시작 노드의 UUID 또는 이름"),
    max_depth: int = Query(5, ge=1, le=20, description="무한 루프 방지를 위한 최대 탐색 깊이"),
    bottom_up: bool = Query(False, description="True=역추적(자식->부모), False=순방향(부모->자식)"),
    client: OccultGrpcClient = Depends(get_grpc_client)
):
    try:
        request = occult_pb2.TraversalRequest(
            start_node_identifier=start_node,
            max_depth=max_depth,
            bottom_up=bottom_up
        )

        response = await client.stub.TraverseGraph(request)

        return {
            "nodes": [serialize_node(n) for n in response.nodes],
            "paths": [serialize_path(p) for p in response.paths]
        }
    except grpc.aio.AioRpcError as e:
        handle_grpc_error(e)

실행 및 테스트

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

~/workspace/occult-graph/python-gateway$ uv run uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

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

python-gateway/tests/test_api.py

import asyncio
import httpx
import json

BASE_URL = "http://localhost:8000/api"

async def test_get_node(client: httpx.AsyncClient):
    print("\n" + "="*50)
    print("🧪 TEST 1: 단건 노드 조회 (GetNode) - Bael")
    print("="*50)

    response = await client.get(f"{BASE_URL}/nodes/Bael")

    if response.status_code == 200:
        data = response.json()
        print(f"✅ 성공! 노드 이름: {data.get('name')}, 타입: {data.get('entity_type')}")
        print(f"📦 속성: {json.dumps(data.get('attributes'), indent=2, ensure_ascii=False)}")
    else:
        print(f"❌ 실패: {response.status_code} - {response.text}")

async def test_search_nodes(client: httpx.AsyncClient):
    print("\n" + "="*50)
    print("🧪 TEST 2: 노드 검색 (SearchNodes) - 'Fire'")
    print("="*50)

    # 쿼리 파라미터 전달
    response = await client.get(f"{BASE_URL}/nodes/search", params={"query": "Fire", "limit": 3})

    if response.status_code == 200:
        data = response.json()
        print(f"✅ 성공! 총 {len(data)}개의 노드를 찾았습니다.")
        for idx, node in enumerate(data, 1):
            print(f"  {idx}. [{node.get('entity_type')}] {node.get('name')}")
    else:
        print(f"❌ 실패: {response.status_code} - {response.text}")

async def test_traverse_graph(client: httpx.AsyncClient):
    print("\n" + "="*50)
    print("🧪 TEST 3: 그래프 재귀 탐색 (TraverseGraph) - '5 of Wands'")
    print("="*50)

    params = {
        "start_node": "5 of Wands",
        "max_depth": 5,
        "bottom_up": "false"
    }
    response = await client.get(f"{BASE_URL}/graph/traverse", params=params)

    if response.status_code == 200:
        data = response.json()
        nodes = data.get("nodes", [])
        paths = data.get("paths", [])
        print(f"✅ 성공! 가져온 노드 수: {len(nodes)}개, 연결 간선 수: {len(paths)}개")

        print("\n[발견된 주요 경로 흐름]")
        # 간선 데이터를 기반으로 관계를 보기 좋게 출력
        for path in paths[:5]: # 너무 많을 수 있으니 상위 5개만 출력
            # ID를 이름으로 변환하기 위한 임시 매핑
            parent_name = next((n["name"] for n in nodes if n["id"] == path["parent_id"]), "Unknown")
            child_name = next((n["name"] for n in nodes if n["id"] == path["child_id"]), "Unknown")
            print(f"  Depth {path['depth']} | {parent_name} --({path['relation_type']})--> {child_name} [가중치: {path['weight']}]")

        if len(paths) > 5:
            print("  ... (이하 생략)")
    else:
        print(f"❌ 실패: {response.status_code} - {response.text}")

async def main():
    print("🚀 FastAPI 게이트웨이 End-to-End 테스트를 시작합니다...")
    async with httpx.AsyncClient() as client:
        await test_get_node(client)
        await test_search_nodes(client)
        await test_traverse_graph(client)
    print("\n🎉 모든 API 테스트가 완료되었습니다.")

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

이 테스트 코드를 실행하기 위해서는 개발의존성을 추가해야 한다.
Rust 엔진 및 FastAPI가 실행 중인 상태로 새 터미널에서 다음을 진행한다.

~/workspace/occult-graph/python-gateway$ uv add --dev httpx
~/workspace/occult-graph/python-gateway$  uv run python tests/test_api.py
🚀 FastAPI 게이트웨이 End-to-End 테스트를 시작합니다...

==================================================
🧪 TEST 1: 단건 노드 조회 (GetNode) - Bael
==================================================
✅ 성공! 노드 이름: Bael, 타입: Demon
📦 속성: {
  "goetia_number": 1,
  "rank": "King"
}

==================================================
🧪 TEST 2: 노드 검색 (SearchNodes) - 'Fire'
==================================================
✅ 성공! 총 3개의 노드를 찾았습니다.
  1. [Element] Fire
  2. [Zodiac] Aries
  3. [Zodiac] Leo

==================================================
🧪 TEST 3: 그래프 재귀 탐색 (TraverseGraph) - '5 of Wands'
==================================================
✅ 성공! 가져온 노드 수: 17개, 연결 간선 수: 16개

[발견된 주요 경로 흐름]
  Depth 1 | 5 of Wands --(BELONGS_TO_ELEMENT)--> Fire [가중치: 1.0]
  Depth 1 | 5 of Wands --(MANIFESTS_IN)--> Geburah of Atziluth [가중치: 1.5]
  Depth 1 | 5 of Wands --(RULES_DECAN)--> Leo Decan 1 [가중치: 2.0]
  Depth 2 | Leo Decan 1 --(PART_OF_SIGN)--> Leo [가중치: 1.0]
  Depth 2 | Geburah of Atziluth --(MANIFESTATION_OF)--> Geburah [가중치: 1.5]
  ... (이하 생략)

🎉 모든 API 테스트가 완료되었습니다.

🤖 AI AGENT | 인라인 메타데이터 vs. 개발의존성

현재처럼 pyproject.toml과 폴더 구조가 갖춰진 정식 애플리케이션 프로젝트의 tests/ 디렉토리 내부에서는 uv add --dev를 사용하여 pyproject.toml에 명시하는 것이 압도적으로 낫습니다.

그 명확한 이유를 실무 아키텍처 관점에서 세 가지로 나누어 설명해 드리겠습니다.

1. 확장성 (Scalability)과 TDD 도입

현재는 test_api.py 하나뿐이지만, 프로젝트가 커지면 test_auth.py, test_graph.py 등으로 테스트 파일이 늘어나고 본격적인 pytest 프레임워크를 도입하게 됩니다.
만약 인라인 메타데이터를 쓴다면 모든 테스트 파일 상단에 의존성 주석을 복사&붙여넣기 해야 하지만, --dev로 관리하면 프로젝트 전체가 공유하는 완벽한 테스트 환경이 구축됩니다.

2. 단일 진실 공급원 (Single Source of Truth)

MSA 환경에서는 이 프로젝트를 처음 인수인계받는 개발자나 CI/CD 파이프라인(GitHub Actions 등)이 오직 pyproject.toml 파일 하나만 보고도 프로젝트의 모든 요구사항을 파악할 수 있어야 합니다.
테스트 라이브러리(httpx, pytest 등)가 파일 주석에 숨어 있으면, 의존성 관리가 파편화되어 유지보수가 매우 어려워집니다.

3. 그렇다면 인라인 메타데이터(/// script)는 언제 쓰나요?

마스터님께서 정확히 활용하셨던 scripts/database/02_data.py 같은 '독립적인 단일 스크립트(Standalone Script)' 에 쓰는 것이 최고의 베스트 프랙티스입니다.
프로젝트 코어 로직과 상관없이 단독으로 실행되며 특정 패키지(psycopg)가 필요할 때, 굳이 가상환경을 파지 않고 스크립트 자체가 의존성을 들고 다니게 만드는 데 최적화된 기능입니다.