FastAPI로 자연어 기반 PostgreSQL 질의하기(1)

FastAPI 사용을 위한 환경 구성 절차

✅ 실험 목적

• AI 에이전트가 자연어로 PostgreSQL에 질문을 던지는 구조를 실험
• FastAPI를 이용해서 간단한 REST API 서버를 구성함으로써, LLM과 DB 사이의 연동 흐름을 손쉽게 체험하고 테스트
• 궁극적으로는 MCP와의 차이를 확인하기 위함

사전 작업 (root 권한)

• 먼저 psycopg2 패키지 빌드를 위해 Python 개발 헤더와 C 컴파일러를 설치

dnf install -y python3-devel gcc

[이하 작업은 postgres 유저로 진행]

1. 실험용 디렉토리 만들기

mkdir /data1/api_pg_test
cd /data1/api_pg_test

2. Python 가상환경(venv) 생성

• 프롬프트 앞에 (venv)가 붙으면 가상환경이 활성화된 상태

python3 -m venv venv
source venv/bin/activate

3. 필요한 패키지 설치

• FastAPI 사용을 위한 패키지 설치

pip install fastapi uvicorn psycopg2

4. FastAPI 서버 코드 작성

vi api_server.py

• python code

from fastapi import FastAPI
from pydantic import BaseModel
import psycopg2
import re
from typing import Any

app = FastAPI()

# PostgreSQL 연결 설정
DB_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "user": "postgres",
    "password": "yourpassword",
    "dbname": "yourdb"
}

class PGRequest(BaseModel):
    input: str

class PGResponse(BaseModel):
    output: Any

# 간단한 프롬프트 → 쿼리 매핑 함수
def prompt_to_query(prompt: str) -> str:
    if "고객" in prompt and "주문" in prompt:
        # 고객 ID 추출 (예: "고객 2" → 2)
        match = re.search(r"고객\s*(\d+)", prompt)
        if match:
            customer_id = match.group(1)
            return f"SELECT * FROM orders WHERE customer_id = {customer_id} ORDER BY order_date DESC LIMIT 5;"
        else:
            return "SELECT '고객 ID를 찾을 수 없습니다.' AS message;"
    else:
        return "SELECT '알 수 없는 요청입니다.' AS message;"

@app.post("/mcp/query", response_model=PGResponse)
async def handle_query(req: PGRequest):
    prompt = req.input
    query = prompt_to_query(prompt)

    try:
        conn = psycopg2.connect(**DB_CONFIG)
        cur = conn.cursor()
        cur.execute(query)
        rows = cur.fetchall()
        colnames = [desc[0] for desc in cur.description]
        result = [dict(zip(colnames, row)) for row in rows]
        cur.close()
        conn.close()
    except Exception as e:
        return PGResponse(output={"error": str(e)})

    return PGResponse(output=result)

@app.get("/")
def root():
    return {"status": "API PostgreSQL 서버 실행 중 🚀"}

5. FastAPI 서버 실행

• 로컬 접속만 허용하는 경우

uvicorn api_server:app --reload

Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

• 외부 접속 허용하는 경우 (서버 IP 바인딩)

• 0.0.0.0으로 설정하면 모든 IP 대역에서 접속 허용 (실험용일 땐 편리하지만, 운영 환경에서는 위험할 수 있음)

uvicorn api_server:app --host 192.x.x.x --port 8000 --reload

6. 로컬 PC에서 접속 테스트

http://192.x.x.x:8000/

• 아래 메세지가 뜨면 정상
  {"status":"API PostgreSQL 서버 실행 중 🚀"}

7. 조회 테스트

• Swagger UI를 통해 API 자연어 질의 테스트

• Request

{
  "input": "고객 2의 주문 내역 보여줘"
}

• Response

{
  "output": [
    {
      "id": 3,
      "customer_id": 2,
      "order_date": "2024-12-03",
      "amount": 20000
    }
  ]
}