DBMS 및 SQL 활용 #1 실습

DBMS 및 SQL 활용 #1 실습 #

#2025-08-27


1. 실습1 #

실습 시나리오

  • 사용자가 설계안 텍스트(예: description)를 입력
  • 해당 텍스트에 대해 Python에서 AI 임베딩을 수행
  • 임베딩 결과가 유효할 경우 design 테이블에 등록 (COMMIT)
  • 실패하면 아무 데이터도 등록하지 않음 (ROLLBACK)
  • PostgreSQL + pgvector 확장 사용
  • Python에서 psycopg2 + 임베딩 처리

#

코드

#1 SQL

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE IF NOT EXISTS design (
    id SERIAL PRIMARY KEY,
    description TEXT,
    embedding VECTOR(1536)  -- OpenAI 임베딩 차원
);

#

#2 python

import psycopg2
import pandas as pd
import os
from dotenv import load_dotenv
from openai import OpenAI

os.chdir("/Users/yshmbid/Documents/home/github/SQL")

# 1. .env 파일 로드
load_dotenv()

# 2. OpenAI Client 생성
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) # .env에서 OPENAI_API_KEY 가져옴

# 3. PostgreSQL DB에 연결
conn = psycopg2.connect(
    host="localhost",
    port=5432,
    database="postgres",
    user="postgres",
    password=os.getenv("PG_PASSWORD"), # .env에서 PG_PASSWORD 가져옴
)
cursor = conn.cursor() # SQL문 실행

# 4. 임베딩 함수
def get_embedding(text: str):
    response = client.embeddings.create(
        input=text,
        model="text-embedding-3-small"  # “text-embedding-3-small” 모델로 임베딩 생성
    )
    return response.data[0].embedding 

# 5. DB 삽입 함수
def insert_design(description: str):
    try:
        cursor.execute("BEGIN;")  # 트랜잭션 시작

        # 임베딩 생성
        embedding = get_embedding(description)

        # design 테이블에 삽입
        cursor.execute(
            "INSERT INTO design (description, embedding) VALUES (%s, %s)",
            (description, embedding)
        )

        conn.commit()
        print(f"[COMMIT] 등록 성공 → {description[:40]}...")

    except Exception as e:
        conn.rollback()
        print(f"[ROLLBACK] 실패 → {description[:40]}... 에러: {e}")

# 6. CSV 파일 로드 & 처리
df = pd.read_csv("sample_designs_500.csv")

for idx, row in df.iterrows():
    desc = row.get("description")
    if pd.notna(desc):  # description이 비어있지 않을 때만 실행
        insert_design(desc)

# 7. 연결 종료
cursor.close()
conn.close()

#

#3 시나리오 구현

  • 사용자가 설계안 텍스트(예: description)를 입력
    • insert_design(desc) -> description(desc): str
  • 해당 텍스트에 대해 Python에서 AI 임베딩을 수행
    • get_embedding(description) -> client.embeddings.create
  • 임베딩 결과가 유효할 경우 design 테이블에 등록 (COMMIT)
    • insert_design -> conn.commit()
  • 실패하면 아무 데이터도 등록하지 않음 (ROLLBACK)
    • insert_design -> except Exception as e -> conn.rollback()

#

#4 개념

트랜젝션?

  • DB에서 여러 SQL 실행을 하나의 작업 단위로 묶는것
  • 여러 SQL 실행?
    • BEGIN; (시작) / INSERT … (데이터 넣기) / UPDATE … (데이터 수정하기) / COMMIT; (끝내기 → 확정 반영) 등 commit?
  • commit 전에는 cursor.execute를 실행해도 DB 내부 버퍼/임시 상태에만 반영됨.
  • commit을 하면 변경사항을 실제 DB 파일(디스크)에 확정 저장되고 다른 클라이언트(psql, pgAdmin 등)에서도 데이터를 조회 가능.

#

2. 실습2 #

실습 시나리오

  • FastAPI 기반 /register_design API를 구현해보세요(Python)
  • Streamlit 를 통해 입력 UI를 만들고 위에 만든 FastAPI를 호출하는 방식으로 해보세요.

아래의 순서대로 진행해보세요.

  1. PostgreSQL의 design 테이블 (생성됨)
  2. FastAPI 서버 실행: uvicorn app:app --reload
  3. Streamlit 클라이언트 실행: streamlit run streamlit_client.py
  4. 입력 → POST → 등록 확인

#

코드

#1 FastAPI 서버

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import psycopg2
from dotenv import load_dotenv
from openai import OpenAI
import os

# 경로 설정
os.chdir("/Users/yshmbid/Documents/home/github/SQL")

# .env 로드
load_dotenv()

# OpenAI 클라이언트
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# DB 연결
conn = psycopg2.connect(
    host="localhost",
    port=5432,
    database="postgres",
    user="postgres",
    password=os.getenv("PG_PASSWORD"),
)
cursor = conn.cursor()

# FastAPI 앱 객체 생성
app = FastAPI()

# 요청 데이터 모델 정의
class DesignRequest(BaseModel):
    description: str

# 임베딩 함수
def get_embedding(text: str):
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

@app.post("/register_design")
def register_design(req: DesignRequest):
    try:
        cursor.execute("BEGIN;")
        embedding = get_embedding(req.description)

        cursor.execute(
            "INSERT INTO design (description, embedding) VALUES (%s, %s)",
            (req.description, embedding)
        )
        conn.commit()
        return {"status": "success", "message": "등록 성공"}
    except Exception as e:
        conn.rollback()
        raise HTTPException(status_code=500, detail=f"등록 실패: {e}")

#2 Streamlit 클라이언트

import streamlit as st
import requests
import os

# 경로 설정
os.chdir("/Users/yshmbid/Documents/home/github/SQL")

st.title("Design 등록 클라이언트")

# 입력 박스
description = st.text_area("설계안 입력", "")

if st.button("등록하기"):
    if description.strip() == "":
        st.warning("설계안을 입력해주세요.")
    else:
        try:
            response = requests.post(
                "http://127.0.0.1:8000/register_design",
                json={"description": description}
            )
            if response.status_code == 200:
                st.success(response.json())
            else:
                st.error(response.json())
        except Exception as e:
            st.error(f"서버 연결 실패: {e}")

#

#3 시나리오 구현

  • FastAPI 서버- class DesignRequest
    • Input: 사용자가 Streamlit 화면에서 입력한 설계안 텍스트를 json {“description”: “텍스트”} 로 변환
    • Pydantic이 json을 검증후 python 객체(req.description)로 변환
    • Output: req.description (문자열)
  • FastAPI 서버- register_design()
    • Input: req.description (문자열)
    • OpenAI API 호출해서 임베딩 벡터 생성 → PostgreSQL design 테이블에 (description, embedding) 저장
    • Output: 성공/실패 메시지 JSON 응답 ({“status”: “success”, “message”: “등록 성공”})
  • Streamlit
    • Input: 사용자가 입력 설계안 description 텍스트
    • FastAPI에 전송하면 json {“description”: “텍스트”} 로 감싸서 fastapi에 POST 요청 → description 데이터 등록
    • Output: 성공 실패 메시지 표시

#

#4 개념

class DesignRequest와 register_design()와의 호환?

  • JSON을 파싱해서 Python 객체로 바꾸고 description이 문자열인지 검증한 뒤 통과하면 register_design()에서 DesignRequest 객체를 만들어 req에 넣는다.

롤백?

  • rollback을 안 하면 “INSERT는 됐는데 commit 전에 에러 발생” 같은 상태가 DB에 남을 수 있음.

#

3. 실습2 - 레퍼런스 코드 #

# AI 설계안 등록 통합 실습 (FastAPI + Streamlit + PostgreSQL)

## FastAPI 서버 (app.py)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer
import psycopg2

app = FastAPI()
model = SentenceTransformer('all-MiniLM-L6-v2')

def get_db_conn():
    return psycopg2.connect(
        dbname="yourdb", user="youruser", password="yourpass", host="localhost"
    )

class DesignInput(BaseModel):
    title: str
    description: str

@app.post("/register_design")
def register_design(data: DesignInput):
    conn = get_db_conn()
    cur = conn.cursor()
    try:
        embedding = model.encode(data.description).tolist()
        cur.execute("BEGIN;")
        cur.execute("""
            INSERT INTO design (title, description, embedding)
            VALUES (%s, %s, %s);
        """, (data.title, data.description, embedding))
        conn.commit()
        return {"status": "success", "message": "등록 완료"}
    except Exception as e:
        conn.rollback()
        raise HTTPException(status_code=500, detail=f"등록 실패: {str(e)}")
    finally:
        cur.close()
        conn.close()
## Streamlit 클라이언트 (streamlit_client.py)
import streamlit as st
import requests

st.title("AI 설계안 등록")

title = st.text_input("설계안 제목")
description = st.text_area("설계안 설명")

if st.button("등록 요청"):
    response = requests.post("http://localhost:8000/register_design", json={
        "title": title,
        "description": description
    })
    if response.status_code == 200:
        st.success(response.json()["message"])
    else:
        st.error(response.json()["detail"])
## 실행 순서 요약
1. PostgreSQL에 `design` 테이블 생성
2. FastAPI 서버 실행: `uvicorn app:app --reload`
3. Streamlit 클라이언트 실행: `streamlit run streamlit_client.py`
4. 입력 → POST → 등록 확인


## 테스트용 CSV & SQL 삽입 예제
- `sample_designs_500.csv`: 샘플 설계안 + 임베딩 포함
- `insert_designs.sql`: INSERT 문 자동 생성

**Tip**: Python에서 `psycopg2`로 임베딩 포함 대량 INSERT 가능

#